Pandas合并CSV文件时部分文件行数暴增百倍的技术问询

原因分析 #

1. 变量名错误：循环读取新文件到file_df，但后续操作的是旧变量file_pri，导致每次循环重复连接同一个历史表，触发笛卡尔积，直接引发行数暴增。
2. NMI非唯一：若原始CSV中NMI存在重复值，设置为索引后进行外连接时，Pandas会对所有匹配的索引行生成笛卡尔积组合，导致行数呈倍数膨胀。
3. 重复设置索引：每次连接后重复执行combined_file.set_index('NMI', inplace=True)，若NMI已作为索引，重复操作会打乱内部索引逻辑，加剧数据异常。

解决方案 #

1. 修正循环变量错误 #

确保循环中操作的是刚读取的新文件数据，替换错误变量名：

counter = 2
for file in csv_files.csv_filename[2:]:
    # 读取当天数据
    file_df = pd.read_csv(file, usecols=['NMI', 'HCT'])
    # 修正为操作file_df而非file_pri
    file_df.set_index('NMI', inplace=True)
    print(file_df.count())
    # 连接时使用file_df
    combined_file = combined_file.join(file_df, 
                                       how='outer', 
                                       rsuffix='_day'+str(dates3[counter]))
    counter +=1
    # 仅在NMI不是索引时执行此操作，避免重复设置
    if 'NMI' in combined_file.columns:
        combined_file.set_index('NMI', inplace=True)

2. 强制保证NMI唯一性 #

在读取每个文件后，检查并清理重复的NMI，避免笛卡尔积：

file_df = pd.read_csv(file, usecols=['NMI', 'HCT'])
# 检查NMI是否重复，保留最后一条（或根据业务选keep='first'）
if not file_df['NMI'].is_unique:
    file_df = file_df.drop_duplicates(subset='NMI', keep='last')
file_df.set_index('NMI', inplace=True)

3. 优化连接逻辑，避免重复操作 #

利用索引直接连接，无需重复指定on='NMI'，减少冗余操作：

# 初始连接后已设置NMI为索引，循环内直接基于索引连接
combined_file = combined_file.join(file_df, how='outer', rsuffix=f'_day{dates3[counter]}')

4. 内存友好的替代方案：concat+groupby #

改用concat合并所有文件，避免多次join的内存损耗：

dfs = []
day_num = 1
for file in csv_files.csv_filename:
    df = pd.read_csv(file, usecols=['NMI', 'HCT'])
    # 去重保证NMI唯一
    df = df.drop_duplicates(subset='NMI', keep='last')
    # 重命名HCT为对应日期
    df.rename(columns={'HCT': f'HCT_day{day_num}'}, inplace=True)
    dfs.append(df)
    day_num +=1

# 按NMI合并所有DataFrame
combined_file = pd.concat([df.set_index('NMI') for df in dfs], axis=1, join='outer')

5. 超大数据分块处理 #

若文件体积过大，采用分块读取+逐块合并，降低内存峰值：

chunk_size = 100000
combined_file = pd.DataFrame()
day_num = 1

for file in csv_files.csv_filename:
    # 分块读取文件
    for chunk in pd.read_csv(file, usecols=['NMI', 'HCT'], chunksize=chunk_size):
        chunk = chunk.drop_duplicates(subset='NMI', keep='last')
        chunk.set_index('NMI', inplace=True)
        chunk.rename(columns={'HCT': f'HCT_day{day_num}'}, inplace=True)
        
        if combined_file.empty:
            combined_file = chunk.copy()
        else:
            combined_file = combined_file.join(chunk, how='outer')
    day_num +=1

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