GroupBy /Map_partitions in Dask

Question

我有一个带有2438个分区的dask数据帧，每个分区是1.1GB，总共7B行我想对多个列进行分组，并聚合其中一列

agg = {'total_x':'sum'}
df_s = df_s.map_partitions(lambda dff: dff.groupby(['y', 'z', 'a', 'b','c']).agg(agg) , meta=pd.DataFrame({'y':'int','z':'int', 'a':'int', 'b':'int','c':'object' ,'total_x':'f64'}))

我得到错误If using all scalar values, you must pass an index

我该如何解决？我有160 GB的RAM和24个工作人员，在那种环境下，这种计算可能吗？

如果没有，还有哪种可行的方法？

Answer 1

正如@Michael Delgado所建议的，meta的定义存在问题。这应该修复meta的定义：

import pandas as pd
dtypes = {
"y": "int",
"z": "int",
"a": "int",
"b": "int",
"c": "object",
"total_x": "f64",
}
meta = pd.DataFrame(columns=dtypes.keys())

然后，这个meta可以作为一个夸尔格传递。参见下面的可复制示例：

import dask
import pandas as pd
dtypes = {"name": "str", "x": "f64"}
meta = pd.DataFrame(columns=dtypes.keys())

agg = {"x": "sum"}
ddf = dask.datasets.timeseries().map_partitions(
lambda df: df.groupby(["name"], as_index=False).agg(agg), meta=meta
)
ddf.head()