best-practices

# Dask 最佳实践 ## 性能优化原则 ### 首先从简单的解决方案开始 在实施 Dask 并行计算之前，先探索以下替代方案： - 针对特定问题的更好算法 - 高效的文件格式（Parquet、HDF5、Zarr 而非 CSV） - 通过 Numba 或 Cython 编译代码 - 用于开发和测试的数据采样 这些替代方案通常比分布式系统提供更好的回报，在扩展到并行计算之前应先穷尽这些方法。 ### 分块大小策略 **关键规则**：块的大小应足够小，以便多个块可以同时放入工作节点的可用内存中。 **推荐目标**：设置块大小，使工作节点每个核心可以容纳 10 个块而不超过可用内存。 **原因**： - 块太大：内存溢出和并行化效率低下 - 块太小：过度的调度开销 **计算示例**： - 8 核，32 GB RAM - 目标：每个块约 400 MB (32 GB / 8 核 / 10 块) ### 使用仪表板进行监控 Dask 仪表板提供了对以下内容的必要可见性： - 工作节点状态和资源利用率 - 任务进度和瓶颈 - 内存使用模式 - 性能特征 访问仪表板以了解并行工作负载中到底是什么变慢了，而不是猜测优化方案。 ## 必须避免的关键陷阱 ### 1. 不要在 Dask 之前在本地创建大对象 **错误做法**： python import pandas as pd import dask.dataframe as dd # 首先将整个数据集加载到内存中 df = pd.read_csv('large_file.csv') ddf = dd.from_pandas(df, npartitions=10) **正确做法**： python import dask.dataframe as dd # 让 Dask 处理加载 ddf = dd.read_csv('large_file.csv') **原因**：先用 pandas 或 NumPy 加载数据会强制调度程序序列化并将这些对象嵌入到任务图中，从而违背了并行计算的目的。 **关键原则**：使用 Dask 方法加载数据，并使用 Dask 控制结果。 ### 2. 避免重复调用 compute() **错误做法**： python results = [] for item in items: result = dask_computation(item).compute() # 每次 compute 都是独立的 results.append(result) **正确做法**： python computations = [dask_computation(item) for item in items] results = dask.compute(*computations) # 一次 compute 处理所有 **原因**：在循环中调用 compute 会阻止 Dask 进行以下操作： - 并行化不同的计算 - 共享中间结果 - 优化