如何解决Dask并行读取大JSON转Parquet的无输出及内存报错问题？

问题根源分析 #

1. 自定义Delayed代码无输出：你遍历Dask DataFrame分区时，仅调用了save(chunk, dest_dir)但未将这些延迟任务收集起来，dask.compute()没有拿到可执行的任务集合，导致任务根本没被触发。
2. 直接to_parquet内存报错：大概率是blocksize设置过大，导致单个块加载后占用内存超出阈值；另外JSON解析的临时开销也可能引发内存溢出。

正确解决方案 #

方案1：修复自定义Delayed任务逻辑 #

需要将每个分区的保存任务收集为延迟对象列表，再统一执行：

import dask
import dask.dataframe as ddf
from dask.delayed import delayed

def save_chunk(chunk, dest_dir, part_num):
    chunk.to_parquet(f"{dest_dir}/part{part_num:02d}.parquet")

def process_large_json(file_path, dest_dir, block_size=100_000_000):
    # 按指定块大小读取JSON生成Dask DataFrame
    ddf_obj = ddf.read_json(
        file_path,
        orient='records',
        lines=True,
        blocksize=block_size
    )
    
    # 收集所有延迟任务
    tasks = []
    for idx, partition in enumerate(ddf_obj.partitions, start=1):
        # 将Dask分区转为Pandas DataFrame（延迟执行）
        pandas_chunk = partition.compute()
        save_task = delayed(save_chunk)(pandas_chunk, dest_dir, idx)
        tasks.append(save_task)
    
    # 并行执行所有任务
    dask.compute(*tasks)

# 调用示例：按100MB分块处理
process_large_json("your_large_data.json", "./parquet_output")

方案2：优化直接to_parquet的写法 #

调整blocksize到合理值，配合Parquet写入参数优化内存占用：

import dask.dataframe as ddf
from dask.config import set

# 配置Dask内存阈值，触发磁盘溢写避免内存爆仓
set({
    "workers.memory.target": "500MB",
    "workers.memory.spill": "700MB"
})

# 按64MB分块读取JSON，写入Parquet
ddf.read_json(
    "your_large_data.json",
    orient='records',
    lines=True,
    blocksize=64_000_000  # 缩小块大小，适配内存
).to_parquet(
    "./parquet_output",
    name_function=lambda i: f'part{i:02d}.parquet',
    write_index=False,  # 不写入索引节省空间
    engine='pyarrow',   # pyarrow引擎对大数据兼容性更优
    overwrite=True
)

关键注意事项 #

• blocksize取值：根据自身内存调整，8GB内存建议设为64MB-128MB，确保单个块加载后不会占满内存。
• 延迟任务管理：必须将所有delayed任务收集到列表，再用dask.compute(*tasks)执行，否则Dask无法识别待执行任务。
• 内存配置：通过Dask的内存阈值设置，让数据在内存不足时自动溢写到磁盘，避免内存错误。

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