大数据场景下合并group_by生成的Series实现内存优化的方案咨询

方案可行性说明 #

你的核心思路是正确的，先对单个文件做预聚合再合并，相比全量拼接原始数据可以降低90%以上的内存占用，完全可以解决2000个大文件的内存溢出问题。你调整后的代码无法运行主要是存在几处语法和逻辑错误，同时要单独处理跨文件同月份数据的合并统计。

原有代码的问题点 #

• 变量引用错误：tempDf['JobDate'] = pd.to_datetime(df['date']) 行中，全局的df在循环中还未赋值，应该引用当前读取的tempDf['date']
• 拼接逻辑错误：每次循环都重新赋值groupConcat，没有累加历史聚合结果，同时ignore_index=True会丢弃groupby生成的state、月份两级索引，导致后续无法合并同维度数据
• 聚合逻辑缺陷：直接对单文件nunique结果拼接后没有二次聚合，无法统计跨文件同月份的合计值；如果同一个id可能出现在多个文件中，直接对单文件nunique结果求和会导致计数偏大

正确实现代码 #

场景1：同一个id仅会出现在单个文件中 #

这种场景下直接单文件预聚合后二次聚合即可，内存占用最低：

# 初始化存储预聚合结果的列表
agg_list = []

for file in files:
    tempDf = pd.read_parquet(file, columns=['state', 'date', 'id'])
    # 转换日期格式，注意引用当前tempDf的date字段
    tempDf['date'] = pd.to_datetime(tempDf['date'])
    # 单文件维度先做分组聚合
    single_file_agg = tempDf.groupby(['state', pd.Grouper(key='date', freq='m')])['id'].nunique().reset_index(name='id_nunique')
    agg_list.append(single_file_agg)

# 合并所有预聚合结果，再做一次分组求和，解决跨文件同月份数据合并问题
total_agg = pd.concat(agg_list, ignore_index=True).groupby(['state', 'date'])['id_nunique'].sum()

# 输出结果
total_agg.to_frame().to_parquet('grouped_states_job_count_total.parquet')

场景2：同一个id可能出现在多个文件中 #

这种场景下不能直接对单文件的nunique结果求和，需要先对单文件的(state,月份,id)去重后再合并，最后做全局去重计数：

agg_list = []

for file in files:
    tempDf = pd.read_parquet(file, columns=['state', 'date', 'id'])
    tempDf['date'] = pd.to_datetime(tempDf['date']).dt.to_period('M')
    # 单文件维度先对(state,月份,id)去重，大幅降低数据量
    single_file_dedup = tempDf[['state', 'date', 'id']].drop_duplicates()
    agg_list.append(single_file_dedup)

# 合并所有去重后的数据，再做全局分组计数
total_df = pd.concat(agg_list, ignore_index=True)
total_agg = total_df.groupby(['state', 'date'])['id'].nunique()

# 输出结果
total_agg.to_frame().to_parquet('grouped_states_job_count_total.parquet')

额外优化建议 #

如果文件数量进一步增大，还可以在循环过程中定期将agg_list的内容落盘临时文件，清空列表后再继续处理，避免预聚合结果本身占用过多内存。

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