如何解决Dask DataFrame合并后索引重复且常规去重方法失效的问题

问题描述 #

合并两个Dask DataFrame后，结果索引出现重复值（示例索引：Index([2, 3, 7, 2, 0, 1, 4, 5, 6, 8, 9, 10, 0, 1], dtype='int64')），对后续计算造成影响。尝试用reset_index()和index.drop_duplicates()去重时触发ValueError，报错信息如下：

ValueError: cannot reindex on an axis with duplicate labels

self = Index([0, 1, 4, 5, 6, 8, 9, 10, 0, 1], dtype='int64')
target = Index([0, 1, 0, 1, 4, 5, 6, 8, 9, 10], dtype='int64'), method = None

合并代码：

def merge_with_aggregated_4(trans_ddf, agg_ddf):
     # 基于分钟调整时间，作为第一次连接条件
    trans_ddf["base_hour"] = trans_ddf["column4"].dt.floor("h")
    trans_ddf['adjusted_hour'] = trans_ddf['column4'].apply(adjust_hour_row, meta=('column4', 'datetime64[ns]'))

    # 第一次连接
    merged_ddf = trans_ddf.merge(agg_ddf, left_on="adjusted_hour", right_on="hour", how="left", suffixes=("", "_joined"))

    # 筛选未匹配的行，移除agg_ddf中除hour外的列
    unmatched_ddf = merged_ddf[merged_ddf["mean_column1"].isna()].drop(
        columns=[col for col in agg_ddf.columns if col != "hour"]
    )

    # 基于原始小时进行二次连接
    unmatched_ddf = unmatched_ddf.merge(agg_ddf, left_on="base_hour", right_on="hour", how="left", suffixes=("", "_fallback"))

    # 用二次连接结果填充缺失值
    for col in agg_ddf.columns:
        if col != "hour":
            merged_ddf[col] = merged_ddf[col].fillna(unmatched_ddf[col])

    return merged_ddf

测试代码：

result_ddf = merge_with_aggregated_4(trans_ddf, agg_ddf)
print(f'结果索引: {result_ddf.index.compute()}')

可行的索引去重方案 #

1. 合并前重置索引（推荐） #

Dask的merge操作会保留原DataFrame的索引，若原索引本身存在重复，合并后会加剧问题。在合并前对两个DataFrame重置索引，从源头避免重复：

def merge_with_aggregated_4(trans_ddf, agg_ddf):
    # 合并前重置索引，丢弃原索引
    trans_ddf = trans_ddf.reset_index(drop=True)
    agg_ddf = agg_ddf.reset_index(drop=True)

    # 后续原有逻辑不变
    trans_ddf["base_hour"] = trans_ddf["column4"].dt.floor("h")
    trans_ddf['adjusted_hour'] = trans_ddf['column4'].apply(adjust_hour_row, meta=('column4', 'datetime64[ns]'))

    merged_ddf = trans_ddf.merge(agg_ddf, left_on="adjusted_hour", right_on="hour", how="left", suffixes=("", "_joined"))

    unmatched_ddf = merged_ddf[merged_ddf["mean_column1"].isna()].drop(
        columns=[col for col in agg_ddf.columns if col != "hour"]
    )

    unmatched_ddf = unmatched_ddf.merge(agg_ddf, left_on="base_hour", right_on="hour", how="left", suffixes=("", "_fallback"))

    for col in agg_ddf.columns:
        if col != "hour":
            merged_ddf[col] = merged_ddf[col].fillna(unmatched_ddf[col])

    # 最终返回前再重置一次索引确保唯一
    return merged_ddf.reset_index(drop=True)

2. 合并后强制生成唯一索引 #

如果无法在合并前处理，可在函数返回结果时直接重置索引，生成新的连续整数索引，彻底解决重复问题：

# 修改函数末尾的返回语句
return merged_ddf.reset_index(drop=True)

3. 保留原索引并去重（需注意数据丢失风险） #

若必须保留原索引信息，可通过分组聚合合并重复索引的行，比如取每组第一行或对数值列做聚合计算：

# 按索引分组，保留每组第一行
deduped_ddf = merged_ddf.groupby(merged_ddf.index).first()

# 或者对数值列取均值（根据业务需求选择聚合方式）
deduped_ddf = merged_ddf.groupby(merged_ddf.index).mean(numeric_only=True)

原报错原因 #

之前调用reset_index()报错，是因为填充缺失值时merged_ddf和unmatched_ddf的索引存在重复，Dask执行fillna时会尝试按索引对齐数据，重复索引导致对齐逻辑冲突。先重置索引再操作即可避免该问题。

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