我想转换我的 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'
试试这个代码:
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 可以处理的方式转换分类列也很重要。