Dask DataFrame groupby 模式（Mode）聚合的实现指南

import pandas as pd
import numpy as np

# 示例数据
data_pandas = pd.DataFrame({
    'status': ['pending', 'pending', 'pending', 'canceled', 'canceled', 'canceled', 'confirmed', 'confirmed', 'confirmed'],
    'clientId': ['A', 'B', 'C', 'A', 'D', 'C', 'A', 'B', 'C'],
    'partner': ['A', np.nan, 'C', 'A', np.nan, 'C', 'A', np.nan, 'C'],
    'product': ['afiliates', 'pre-paid', 'giftcard', 'afiliates', 'pre-paid', 'giftcard', 'afiliates', 'pre-paid', 'giftcard'],
    'brand': ['brand_4', 'brand_2', 'brand_3', 'brand_1', 'brand_2', 'brand_3', 'brand_1', 'brand_3', 'brand_3'],
    'gmv': [100, 100, 100, 100, 100, 100, 100, 100, 100]
})

data_pandas = data_pandas.astype({
    "partner": "category",
    "status": "category",
    "product": "category",
    "brand": "category"
})

# 使用 Pandas 计算分组模式
mode_pandas = data_pandas.groupby(["clientId", "product"], observed=True).agg({"brand": pd.Series.mode})
print("Pandas Groupby Mode Result:")
print(mode_pandas)

import dask.dataframe as dd
from dask.dataframe import Aggregation

# 将 Pandas DataFrame 转换为 Dask DataFrame
df_dask = dd.from_pandas(data_pandas, npartitions=1) # npartitions=1 简化示例，实际应用中可根据数据大小调整

# 定义自定义的 Dask 模式聚合
mode_dask_agg = Aggregation(
    name="mode", # 聚合的名称
    chunk=chunk,
    agg=agg,
    finalize=finalize,
)

# 应用自定义聚合
mode_dask_result = df_dask.groupby(["clientId", "product"], observed=True, dropna=True).aggregate(
    {"brand": mode_dask_agg}
).compute() # .compute() 触发计算并返回 Pandas DataFrame

print("\nDask Groupby Mode Result:")
print(mode_dask_result)

from pandas import DataFrame, Series, NA
import pandas as pd
from dask.dataframe import from_pandas, Aggregation
import dask.dataframe as dd
import numpy as np

# 1. 准备数据
data = DataFrame(
    {
        "status": [
            "pending", "pending", "pending", "canceled", "canceled", "canceled", "confirmed", "confirmed", "confirmed",
        ],
        "clientId": ["A", "B", "C", "A", "D", "C", "A", "B", "C"],
        "partner": ["A", NA, "C", "A", NA, "C", "A", NA, "C"],
        "product": [
            "afiliates", "pre-paid", "giftcard", "afiliates", "pre-paid", "giftcard", "afiliates", "pre-paid", "giftcard",
        ],
        "brand": [
            "brand_4", "brand_2", "brand_3", "brand_1", "brand_2", "brand_3", "brand_1", "brand_3", "brand_3",
        ],
        "gmv": [100, 100, 100, 100, 100, 100, 100, 100, 100],
    }
)

data = data.astype(
    {
        "partner": "category",
        "status": "category",
        "product": "category",
        "brand": "category",
    }
)

# 2. Pandas 模式聚合（作为对比）
mode_pandas = data.groupby(["clientId", "product"], observed=True).agg(
    {"brand": Series.mode}
)
print("Pandas Groupby Mode Result:")
print(mode_pandas)

# 3. 转换为 Dask DataFrame
df_dask = from_pandas(data, npartitions=1)

# 4. 定义 Dask 自定义聚合函数的三个阶段
def chunk(s):
    """在每个 Dask 分区上执行，计算每个值的频率。"""
    return s.value_counts()

def agg(s0):
    """合并来自不同分区的中间结果（频率计数）。"""
    _intermediate = s0._selected_obj.groupby(level=s0._selected_obj.index.names).sum()
    _intermediate = _intermediate[_intermediate > 0]
    return _intermediate

def finalize(s):
    """从合并后的频率计数中确定最终的众数。"""
    level = list(range(s.index.nlevels - 1))
    return s.groupby(level=level, group_keys=False).apply(lambda x: x[x == x.max()])

# 5. 创建 dask.dataframe.Aggregation 对象
mode_dask_agg = Aggregation(
    name="mode",
    chunk=chunk,
    agg=agg,
    finalize=finalize,
)

# 6. 在 Dask DataFrame 上应用自定义聚合
mode_dask_result = df_dask.groupby(["clientId", "product"], observed=True, dropna=True).aggregate(
    {"brand": mode_dask_agg}
).compute()

print("\nDask Groupby Mode Result:")
print(mode_dask_result)