Dask拼接重分区大体积时序parquet文件相关性分析问题

• 数据集属性：跨度11年的秒级采样时序数据，共约100个数据列，索引为Pandas to_datetime() 生成的datetime类型时间序列
• 业务目标：开展列间相关性分析（单次计算仅需加载2列即可），支持按48秒、1小时、月等不同时间粒度重采样，最终实现11年全周期的相关性结果可视化
• 原始存储：数据按年度拆分为11个独立Parquet文件，由Pandas从原始txt文件生成，写入时未做分区处理，单文件全量加载到内存约占20GB
• 硬件限制：目标运行机器内存仅16GB，实测加载11年跨度的单列数据需占用约10GB内存，无法同时加载两列数据完成计算

用户最初选择Dask替代Pandas适配低内存环境，计划完成两个操作：

1. 拼接所有年度Parquet文件
2. 将拼接后的数据重分区为合适数量的分片，实现仅加载2列时不会占满内存
   初始执行代码如下：

# 读取目录下所有11个Parquet文件
df = dd.read_parquet("/blah/parquet/", engine='pyarrow')
# 重分区为20个Parquet文件导出
df.repartition(npartitions=20).to_parquet("/mnt/data2/SDO/AIA/parquet/combined")

执行第二步重分区写入Parquet时，内存占用飙升直接触发内核关闭，不符合Dask支持超内存数据集计算的预期。

后续用户用Pandas重新生成Parquet文件，将单文件的行组数量从默认的1个调整为约20个，此时出现新问题：

• 无论split_row_groups参数设置为True还是False，都无法直接在Dask中执行重采样操作（例如myseries = myseries.resample('48s').mean()）
• 必须先对Dask序列调用compute()转换为Pandas对象才能运行重采样，完全违背Dask分块处理数据的设计初衷
  执行重采样时触发如下报错：

ValueError: Can only resample dataframes with known divisions
更多说明可参考Dask官方数据框设计文档的分区章节
该问题在最初单Parquet文件仅1个行组的配置下不会出现。

故障根因 #

1. 重分区OOM：divisions（分区边界）未知时，repartition(npartitions=20)会触发全量数据shuffle，需要把所有数据加载到内存完成分块，11年全量数据总大小超200GB，16GB内存必然崩溃
2. 重采样报错：拆分多行列组后，Dask无法自动识别每个行组的时间索引边界，判定分区为未知状态，而Dask的resample接口要求必须已知分区边界才能执行分块计算

操作步骤 #

• 步骤1：手动绑定已知时间分区边界，解决重采样报错
  原始数据按年度拆分、且时间索引有序，不需要全量扫描数据推断分区，直接手动传入年度时间边界即可，执行完成后可直接在Dask对象上调用resample，无需提前compute()：

  import dask.dataframe as dd
  import pandas as pd
  # 根据实际数据的起止年份生成年度时间边界，例：数据覆盖2010-2020年共11年
  year_divisions = pd.date_range(start="2010-01-01", end="2021-01-01", freq="AS").tolist()
  # 读取数据时关闭自动分区推断，避免全量扫描占内存
  df = dd.read_parquet(
      "/blah/parquet/",
      engine="pyarrow",
      index="timestamp", # 替换为实际的时间索引列名
      split_row_groups=True,
      calculate_divisions=False
  )
  # 绑定已知分区边界，标记索引已排序
  df = df.set_index("timestamp", divisions=year_divisions, sorted=True)
  

• 步骤2：按时间频率重分区，避免shuffle OOM
  不要用固定分区数的重分区方式，改为按时间粒度重分区，由于数据已按时间排序，该操作仅需按时间边界切分现有块，几乎无额外内存开销。重分区粒度选择单分片覆盖1个月即可，11年共132个分片，单分片加载2列的内存占用仅约150MB，远低于内存上限。写入前先配置本地Dask集群限制单worker内存，避免单任务占满资源：

  from dask.distributed import Client, LocalCluster
  # 启动本地集群，限制单worker内存为4GB
  cluster = LocalCluster(n_workers=2, memory_limit='4GB', threads_per_worker=2)
  client = Client(cluster)
  # 按月度时间粒度重分区，无全量shuffle
  df = df.repartition(freq="1M")
  # 写入时控制行组大小，单行列组控制在10万行（秒级数据约对应1.15天）
  df.to_parquet(
      "/mnt/data2/SDO/AIA/parquet/combined",
      engine="pyarrow",
      row_group_size=100_000,
      write_index=True,
      overwrite=True
  )
  

• 步骤3：计算阶段按需加载列，进一步降低内存占用
  做相关性计算时，仅读取需要的2个列，不要加载全量字段：

  # 仅读取目标两列，其余98列不会被加载到内存
  df_sub = dd.read_parquet(
      "/mnt/data2/SDO/AIA/parquet/combined",
      engine="pyarrow",
      columns=["col_a", "col_b"],
      index="timestamp"
  )
  # 直接执行多粒度重采样，无需提前compute
  resampled_48s = df_sub.resample("48s").mean().compute()
  resampled_1h = df_sub.resample("1h").mean().compute()
  resampled_1m = df_sub.resample("1M").mean().compute()
  # 后续直接开展相关性计算、可视化即可
  corr_val = resampled_48s["col_a"].corr(resampled_48s["col_b"])
  

内容的提问来源于stack exchange，提问作者Wall-E