如何用Dask从多Parquet文件创建多级DataFrame实现高效绘图

完全可以用Dask实现你的需求，通过并行读取Parquet文件构建多级索引Dask DataFrame，大幅提升数据加载和预处理速度，最终实现和Pandas相同的绘图效果。以下是具体实现步骤和代码：

核心实现思路 #

1. 并行批量读取Parquet：利用Dask的多进程能力同时读取所有Parquet文件，为每个文件添加唯一标识（对应原文件名）作为多级索引的第一级。
2. 构建多级索引结构：合并所有文件数据为统一的Dask DataFrame，索引结构为(file_id, t)，和你之前Pandas实现的多级索引逻辑一致。
3. 按需计算并转Pandas：仅提取需要绘图的参数列，通过Dask并行完成unstack操作后转为内存中的Pandas DataFrame（Matplotlib需要内存数据绘图）。
4. 绘制独立曲线：基于处理后的数据绘制所有测试结果的参数变化曲线。

完整代码示例 #

import os
import dask.dataframe as dd
import matplotlib.pyplot as plt

# 1. 获取所有Parquet文件路径
results_dir = "results"
parquet_files = [
    os.path.join(results_dir, f) 
    for f in os.listdir(results_dir) 
    if f.endswith(".parquet")
]

# 2. 定义读取函数：为每个文件添加file_id作为索引第一级
def read_parquet_with_id(file_path):
    # 仅读取必要列（t和目标参数，这里以reg为例），减少数据加载量
    df = dd.read_parquet(file_path, columns=["t", "reg"])
    # 提取文件名（去掉.parquet后缀）作为测试结果标识
    file_id = os.path.splitext(os.path.basename(file_path))[0]
    # 设置多级索引：(file_id, t)
    df = df.assign(file_id=file_id).set_index(["file_id", "t"])
    return df

# 3. 并行合并所有文件为Dask多级索引DataFrame
combined_dask = dd.concat([read_parquet_with_id(f) for f in parquet_files], axis=0)

# 4. 提取目标参数，并行完成unstack后转为Pandas DataFrame
# unstack后每个file_id对应一列，t作为行索引
reg_data = combined_dask["reg"].unstack(level=0).compute()

# 5. 绘制所有独立曲线
plt.figure(figsize=(12, 6))
plt.plot(reg_data)
plt.xlabel("时间t")
plt.ylabel("reg参数值")
plt.title("所有测试结果中reg随时间的变化")
plt.show()

关键优化点 #

• 列裁剪：读取Parquet时指定columns参数，只加载t和目标参数列，避免无关数据占用内存，大幅提升读取速度。
• 并行计算：Dask自动将读取、合并、unstack等任务分配到多个CPU核心并行执行，比单线程Pandas效率提升数倍。
• 延迟计算：Dask在调用compute()前仅构建计算逻辑图，不会加载全量数据，确保只执行必要的计算步骤。

适配未知参数的场景 #

如果无法提前确定要绘制的参数，可先读取单个文件获取列名，再动态选择目标参数：

# 获取单个文件的列名，确定可选参数
sample_df = dd.read_parquet(parquet_files[0])
available_params = [col for col in sample_df.columns if col != "t"]

# 动态选择目标参数（比如用户指定绘制"temp"）
target_param = "temp"

# 重新定义读取函数，加载t和目标参数
def read_parquet_with_target(file_path):
    df = dd.read_parquet(file_path, columns=["t", target_param])
    file_id = os.path.splitext(os.path.basename(file_path))[0]
    df = df.assign(file_id=file_id).set_index(["file_id", "t"])
    return df

combined_dask = dd.concat([read_parquet_with_target(f) for f in parquet_files], axis=0)
param_data = combined_dask[target_param].unstack(level=0).compute()

plt.plot(param_data)
plt.xlabel("时间t")
plt.ylabel(f"{target_param}参数值")
plt.show()

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