尝试修改分类数据时出现 dask 错误

Question

我想转换我的 dask.DataFrame 的一个分类列。

这就是我现在正在做的事情：

# toy example
df = pd.DataFrame({'file': ['A.csv', 'B.csv', 'C.csv']})
df['file'] = df.file.astype('category')
ddf = dd.from_pandas(df, npartitions=2)

# get new category
ddf["id"] = ddf["file"].cat.rename_categories(lambda x : x.split('.')[0])

当我检查

ddf["id"].cat.categories

时，结果正如预期的那样：

Index(['A', 'B', 'C'], dtype='object')

但是，

ddf.compute()

会引发以下错误：

TypeError: <lambda>() missing 1 required positional argument: 'x'

Answer 1

试试这个代码：

import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({'file': ['A.csv', 'B.csv', 'C.csv']})
df['file'] = df['file'].astype('category')
ddf = dd.from_pandas(df, npartitions=2)

# Define named function for transformation
def extract_id_from_file(df):
    df['id'] = df['file'].str.split('.').str[0]
    return df

# Apply transformation using map_partitions
ddf = ddf.map_partitions(extract_id_from_file)

# Inspect new categories
print(ddf['id'].compute().unique())

# Compute dask DF
result = ddf.compute()
print(result)

事实是，Dask 无法在分布式设置中序列化 lambda 函数。必须使用命名函数而不是 lambda 函数。

通过使用 Dask 的“map_partitions”映射类别转换，以 Dask 可以处理的方式转换分类列也很重要。