对许多镶木地板文件进行 Dask 数据帧重新洗牌

Question

我有一个分散在许多工作节点周围的 dask 集群。 我还有一个 S3 存储桶，其中包含尽可能多的 parquet 文件（现在 500k 文件，将来可能是三倍）。

parquet 中的数据主要是文本： [用户名、名字、姓氏、电子邮件、电子邮件域]

我想加载它们、重新排列它们并存储新的分区。 我希望能够根据 email_domain 对操作进行分组，所以我想为每个 email_domain 分组一个新的 parquet 文件

目前我使用 from_delayed 和 groupby，但生成的 DAG 有一个 shuffle-split 层，大小为 n**2 - 这不适合我的调度程序内存。 类似的东西：

def store(x):
   path = f's3://bucket/{x.name}.parquet'
   x.to_parquet(path)
   return path

z = df.groupby('email_domain').apply(store, meta=('email_domain', 'object'))
visualize(z)
z.compute()

Answer 1

是的，groupby-apply 很昂贵，尤其是并行。

我希望事情仍然有效，但速度会很慢。