Блог

Я пытаюсь запустить функцию во многих разделах фрейма данных Dask. Код требует распаковки кортежей и хорошо работает с Pandas, но не с Dask map_partitions . Данные соответствуют спискам кортежей, где длина списков может варьироваться, но кортежи всегда имеют известную фиксированную длину.

 import dask.dataframe as dd
import pandas as pd

def func(df):
    for index, row in df.iterrows():
        tuples = row['A']
        for t in tuples:
            x, y = t
          # Do more stuff

# Create Pandas dataframe
# Each list may have a different length, tuples have fixed known length
df = pd.DataFrame({'A': [[(1, 1), (3, 4)], [(3, 2)]]})
# Pandas to Dask
ddf = dd.from_pandas(df, npartitions=2)

# Run function over Pandas dataframe
func(df)
# Run function over Dask dataframe
ddf.map_partitions(func).compute()
 

Здесь версия Pandas работает без проблем. Однако Dask вызывает ошибку:

 ValueError: Metadata inference failed in `func`.

You have supplied a custom function and Dask is unable to 
determine the type of output that that function returns. 

To resolve this please provide a meta= keyword.
The docstring of the Dask function you ran should have more information.

Original error is below:
------------------------
ValueError('not enough values to unpack (expected 2, got 1)')
 

В моей исходной функции я использую эти кортежи в качестве вспомогательных переменных, и данные, которые в конечном итоге возвращаются, совершенно разные, поэтому использование meta не решает проблему. Как я могу распаковать кортежи?

Когда вы используете map_partitions без указания meta , dask попытается запустить функции, чтобы определить, каков результат. Это может вызвать проблемы, если ваша функция несовместима с используемым образцом фрейма данных, вы можете увидеть этот пример фрейма данных с ddf._meta_nonempty помощью (в этом случае он вернет столбец foo ).

Простым решением в этом случае является предоставление meta , чтобы возвращаемые данные были разного формата, например, если каждый возвращаемый результат представляет собой список, вы можете предоставить meta=list :

 import dask.dataframe as dd
import pandas as pd

def func(df):
    for index, row in df.iterrows():
        tuples = row['A']
        for t in tuples:
            x, y = t
    return [1,2,3]

df = pd.DataFrame({'A': [[(1, 1), (3, 4)], [(3, 2)]]})
ddf = dd.from_pandas(df, npartitions=2)
ddf.map_partitions(func, meta=list).compute()
 

Другой подход заключается в том, чтобы сделать вашу функцию совместимой с используемым образцом фрейма данных. Пример dataframe имеет столбец object, но он содержит foo , а не список кортежей, поэтому его нельзя распаковать как кортеж. Изменение вашей функции для приема столбцов, отличных от кортежей (с x, *y = t ), заставит ее работать:

 import dask.dataframe as dd
import pandas as pd

def func(df):
    for index, row in df.iterrows():
        tuples = row['A']
        for t in tuples:
            x, *y = t
    return [1,2,3]

df = pd.DataFrame({'A': [[(1, 1), (3, 4)], [(3, 2)]]})
ddf = dd.from_pandas(df, npartitions=2)
#notice that no meta is specified here
ddf.map_partitions(func).compute()