Spark：将键值对数据转换为列格式的技术需求

嘿，这个场景我太熟悉了！要把十万个长格式的小文件转成以itemID为行、key为列的宽表DataFrame，其实用pandas就能搞定，但得注意十万文件的规模，不能硬怼内存。下面给你一步步拆解：

本质上这是长表转宽表的典型需求：把每个itemID对应的多条(key, value)记录，合并成一行多列的结构。因为文件数量极多，必须考虑分批/并行处理，避免内存溢出。

1. 先写单个文件的处理函数 #

先搞定单个文件的转换逻辑，确保每个文件能输出符合要求的小DataFrame：

import pandas as pd
import os

def process_single_file(file_path):
    # 读取文件：因为没有表头，手动指定列名
    df = pd.read_csv(
        file_path, 
        header=None, 
        names=['itemID', 'key', 'value'],
        dtype={'itemID': str, 'key': str}  # 指定数据类型，减少内存占用
    )
    
    # 长表转宽表：如果同一个itemID+key有重复值，用pivot_table替代pivot，指定聚合逻辑
    # 比如取第一个值、均值，根据你的业务需求调整
    pivoted_df = df.pivot_table(
        index='itemID', 
        columns='key', 
        values='value', 
        aggfunc='first'  # 遇到重复取第一个值
    ).reset_index()
    
    return pivoted_df

2. 批量处理十万个文件 #

考虑到文件数量大，这里提供两种方案：

方案一：串行分批处理（适合内存有限的机器）

def batch_process_serial(folder_path):
    # 获取目标文件夹下所有文件路径
    file_paths = [
        os.path.join(folder_path, f) 
        for f in os.listdir(folder_path) 
        if os.path.isfile(os.path.join(folder_path, f))
    ]
    
    # 初始化结果容器
    final_df = pd.DataFrame()
    
    for idx, file_path in enumerate(file_paths):
        if idx % 1000 == 0:
            print(f"已处理 {idx} 个文件...")
        try:
            single_df = process_single_file(file_path)
            # 用outer join合并，保留所有itemID和key
            final_df = pd.merge(final_df, single_df, on='itemID', how='outer')
        except Exception as e:
            print(f"处理文件 {file_path} 出错：{str(e)}")
            continue
    
    # 清理列名：合并时重复key会加_x/_y后缀，这里统一去掉
    final_df.columns = [col.split('_')[0] if '_' in col else col for col in final_df.columns]
    # 填充缺失值，根据业务需求换成0/NaN等
    final_df = final_df.fillna('')
    
    return final_df

方案二：并行处理（加快速度，适合多核机器）

十万个文件串行处理太慢，用多进程并行处理能大幅提升效率：

from concurrent.futures import ProcessPoolExecutor

def batch_process_parallel(folder_path, max_workers=4):
    file_paths = [
        os.path.join(folder_path, f) 
        for f in os.listdir(folder_path) 
        if os.path.isfile(os.path.join(folder_path, f))
    ]
    
    # 并行处理所有文件
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        processed_dfs = list(executor.map(process_single_file, file_paths))
    
    # 合并所有结果：先拼接再按itemID去重
    final_df = pd.concat(processed_dfs, ignore_index=True).groupby('itemID').first().reset_index()
    # 填充缺失值
    final_df = final_df.fillna('')
    
    return final_df

• 内存优化：读取文件时指定dtype，避免pandas自动推断类型占用额外内存；如果value是数值型，直接指定float/int
• 增量保存：如果最终结果太大无法存入内存，可以每处理1000个文件就把中间结果append到Parquet/CSV文件（Parquet比CSV更节省空间）
• 重复值处理：如果同一个itemID+key在多个文件出现，一定要明确aggfunc的逻辑（取第一个、均值、求和等），避免数据混乱

假设某文件内容为：

item1_1,key1,value1
item1_1,key2,value2
item1_2,key1,value3

处理后会得到：

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