处理60万条含数组的JSON：PySpark拆分合并性能优化咨询

核心流程梳理 #

• 读取JSON文件生成初始DataFrame
• 展平past_features/current_features/future_features三个数组
• 拆分出对应子DataFrame并添加feature_type标识列
• 合并所有子DataFrame并统计各部分行数
• 将最终结果写入CSV文件

关键性能优化建议 #

1. JSON读取阶段 #

• 采用行式解析：如果JSON是每行一条记录的格式，使用pandas.read_json(..., lines=True)，比默认的整体解析速度提升明显
• 分块读取：通过chunksize参数（如chunksize=10000）拆分文件，避免一次性加载60万条记录到内存
• 使用PyArrow引擎：Pandas 2.0+支持engine='pyarrow'，解析速度远快于默认引擎，尤其适合大规模数据

2. 数组展平与拆分阶段 #

• 利用内置高效方法：用explode()展平数组，替代自定义循环；用pd.json_normalize()一次性解析数组内的JSON对象，避免逐个字段提取
• 分块处理减少内存压力：在每个数据块内完成展平、添加标识、合并操作，再汇总所有块的结果
  示例代码：

  import pandas as pd
  
  chunks = []
  # 分块读取JSON
  for chunk in pd.read_json('data.json', lines=True, chunksize=10000, engine='pyarrow'):
      # 处理past_features
      past_df = chunk.explode('past_features').reset_index(drop=True)
      past_df = pd.concat([past_df.drop('past_features', axis=1), pd.json_normalize(past_df['past_features'])], axis=1)
      past_df['feature_type'] = 'past'
      
      # 处理current_features
      current_df = chunk.explode('current_features').reset_index(drop=True)
      current_df = pd.concat([current_df.drop('current_features', axis=1), pd.json_normalize(current_df['current_features'])], axis=1)
      current_df['feature_type'] = 'current'
      
      # 处理future_features
      future_df = chunk.explode('future_features').reset_index(drop=True)
      future_df = pd.concat([future_df.drop('future_features', axis=1), pd.json_normalize(future_df['future_features'])], axis=1)
      future_df['feature_type'] = 'future'
      
      # 合并当前块的三个DataFrame
      chunks.append(pd.concat([past_df, current_df, future_df], ignore_index=True))
      # 释放当前块的中间变量内存
      del past_df, current_df, future_df
  
  # 合并所有块得到最终结果
  final_df = pd.concat(chunks, ignore_index=True)
  

3. 内存占用优化 #

• 指定数据类型：读取时为列设置合适的dtype，减少内存消耗，比如：

  dtype_spec = {
      'id': 'string',
      'as_of_date': 'datetime64[ns]',
      'smartdevice': 'boolean'
  }
  pd.read_json('data.json', lines=True, dtype=dtype_spec, engine='pyarrow')
  

• 及时清理中间数据：处理完每个块后，删除临时DataFrame并调用gc.collect()（需导入gc模块）释放内存
• 避免冗余变量：尽量使用链式操作，减少不必要的中间变量创建

4. 行数统计优化 #

• 分块累加统计：无需展平即可提前统计各数组的行数，节省时间和内存：

  total_past = 0
  total_current = 0
  total_future = 0
  
  for chunk in pd.read_json('data.json', lines=True, chunksize=10000, engine='pyarrow'):
      total_past += chunk['past_features'].apply(len).sum()
      total_current += chunk['current_features'].apply(len).sum()
      total_future += chunk['future_features'].apply(len).sum()
  
  print(f"Past Features行数: {total_past}")
  print(f"Current Features行数: {total_current}")
  print(f"Future Features行数: {total_future}")
  print(f"合并后总行数: {total_past + total_current + total_future}")
  

5. CSV写入阶段 #

• 分块写入：使用to_csv(..., chunksize=100000)避免一次性写入超大文件
• 关闭索引：设置index=False减少文件体积和写入时间
• 使用PyArrow引擎：engine='pyarrow'大幅提升写入速度

进阶优化方向 #

• 并行处理：利用swifter库或Dask框架，借助多CPU核心加速数据处理
• 分布式计算：对于超大规模数据，使用Dask DataFrame替代Pandas，支持分布式内存处理，避免单机内存瓶颈

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