如何使用Dask解析嵌套JSON文件并转换为指定结构DataFrame

前置依赖 #

首先安装需要的运行库：

pip install dask[complete] pandas numpy

通用初始化步骤 #

处理逻辑基于Dask Bag实现，适合GB级半结构化JSON的并行处理：

import dask.bag as db
import json
import pandas as pd
import numpy as np

# 读取JSON文件，blocksize按可用内存调整，建议设为可用内存的1/10左右，示例为128MB分块
# 适用场景：文件每行存储一个完整JSON对象（GB级JSON最常用的存储格式）
json_bag = db.read_text("your_data_file.json", blocksize="128MB").map(json.loads)

# 提取data下的result数组，压平为每个result元素为单独的处理单元
result_bag = json_bag.pluck("data").pluck("result").flatten()

如果你的文件是单个超大JSON对象而非行存JSON，可先用ijson库迭代提取data.result下的所有元素，再转为Dask Bag处理，避免加载整个文件到内存。

需求1：生成保留完整values数组的DataFrame #

实现逻辑为展开metric字段为列，缺失字段自动补NaN，保留完整values数组：

def map_metric_with_values(item):
    res = item["metric"].copy()
    res["values"] = item["values"]
    return res

# 转为Dask DataFrame，自动推断所有metric字段，缺失值补NaN
df1 = result_bag.map(map_metric_with_values).to_dataframe()

# 结果输出：建议直接写Parquet避免爆内存，不需要调用compute加载到本地
# df1.to_parquet("output_df1", write_index=False)
# 如需查看小批量结果可以取前几行：df1.head(10)

输出结构与要求的第一种格式完全一致。

需求2：拆分values为独立行的DataFrame #

实现逻辑为展开values数组的每一项为单独行，关联对应metric字段：

def flatmap_metric_explode_values(item):
    metric = item["metric"]
    for time_val, value_val in item["values"]:
        row = metric.copy()
        row["time"] = time_val
        row["value"] = value_val
        yield row

# 压平生成的多行结果，转为Dask DataFrame
df2 = result_bag.flatmap(flatmap_metric_explode_values).to_dataframe()
# 调整列顺序，将time、value放在最前
df2 = df2[["time", "value"] + [col for col in df2.columns if col not in ["time", "value"]]]

# 结果输出：同样建议直接写Parquet
# df2.to_parquet("output_df2", write_index=False)

输出结构与要求的第二种格式完全一致。

优化建议 #

• 若提前已知所有metric字段，可在to_dataframe时指定meta参数，避免Dask采样推断字段的开销，示例：

# 需求1的meta示例
meta1 = {
    "data0": str, "data1": str, "data2": str, "data3": str,
    "values": object
}
df1 = result_bag.map(map_metric_with_values).to_dataframe(meta=meta1)

• 处理TB级数据时，可开启Dask分布式集群，进一步提升并行效率。
• 避免直接调用compute()将全量数据加载到本地内存，优先写为Parquet等列式存储格式供后续分析使用。

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