如何在Dask的groupby聚合中实现拼接唯一字符串的Lambda函数？

Dask的groupby.agg不支持匿名lambda作为聚合操作，核心原因是分布式环境下无法序列化匿名函数分发到各个分区执行。要实现和Pandas相同的功能，可采用以下两种方案：

方法一：分两步处理（先取唯一值再拼接） #

先通过分组获取每组的唯一值列表，再对列表执行字符串拼接，注意需指定meta参数明确结果类型：

import pandas as pd
import dask.dataframe as dd

# 创建原始数据
A = pd.DataFrame(
    data={
        "A": ["saad", "saad", "saad", "saad", "nimra", "asad", "nimra", "nimra", "asad"],
        "B": ["hello", "hello", "saad", "whatsup?", "yup", "nup", "saad", "saad", "nup"],
        "C": ["hello", "hello", "saad", "whatsup?", "yup", "nup", "saad", "saad", "nup"]
    }
)

# 转换为Dask DataFrame
dask_df = dd.from_pandas(A, npartitions=2)

# 1. 分组获取每个字段的唯一值列表
grouped_unique = dask_df.groupby("A")[["B", "C"]].unique()

# 2. 对每个分组的唯一值列表执行拼接
result = grouped_unique.apply(
    lambda x: {col: ', '.join(x[col]) for col in x.index},
    axis=1,
    meta=pd.Series(dtype=str, index=["B", "C"])
)

# 计算并输出结果
print(result.compute())

方法二：使用命名聚合函数 #

定义可序列化的命名函数，直接传入agg方法，同时指定meta参数：

import pandas as pd
import dask.dataframe as dd

# 创建原始数据
A = pd.DataFrame(
    data={
        "A": ["saad", "saad", "saad", "saad", "nimra", "asad", "nimra", "nimra", "asad"],
        "B": ["hello", "hello", "saad", "whatsup?", "yup", "nup", "saad", "saad", "nup"],
        "C": ["hello", "hello", "saad", "whatsup?", "yup", "nup", "saad", "saad", "nup"]
    }
)

# 定义命名聚合函数
def concat_unique(series):
    return ', '.join(series.unique())

# 转换为Dask DataFrame并执行聚合
dask_df = dd.from_pandas(A, npartitions=2)
result = dask_df.groupby("A").agg(
    {
        "B": concat_unique,
        "C": concat_unique
    },
    meta={"B": str, "C": str}
)

# 计算并输出结果
print(result.compute())

关键注意事项 #

• Dask无法序列化匿名lambda函数，必须使用命名函数或可序列化的函数对象。
• 必须指定meta参数，帮助Dask推断结果的数据结构和类型，避免运行时错误。

内容的提问来源于stack exchange，提问作者Saad Munir