Dask 是一个用于并行与分布式计算的 Python 库，可将 pandas、NumPy 等操作扩展到超出内存的数据集。它提供三项关键能力：在单机上对超过可用 RAM 的数据执行“大于内存”的计算；跨多核并行处理以提升计算速度；支持跨多台机器处理 TB 级数据集的分布式计算。Dask 的规模可从笔记本（约处理 100 GiB）扩展到集群（约处理 100 TiB），同时保持熟悉的 Python API。
适用场景 当需要处理超过可用 RAM 的数据集、将 pandas 或 NumPy 操作扩展到更大数据、并行化计算以提升性能、高效处理多个文件（CSV、Parquet、JSON、文本日志）、构建带任务依赖的自定义并行工作流，或在多核或多台机器间分布工作负载时，可使用本技能。
核心组件 Dask 提供五个主要组件，各自适合不同用例：
- DataFrames：并行 pandas 操作 用途：通过并行处理将 pandas 操作扩展到更大数据集。适用于表格数据超过可用 RAM、需要同时处理多个 CSV/Parquet 文件、pandas 操作缓慢需要并行化、从 pandas 原型扩展到生产环境等场景。 示例： import dask.dataframe as dd
将多个文件读取为单个 DataFrame
ddf = dd.read_csv('data/2024-*.csv')
在调用 compute() 之前，操作是惰性的
filtered = ddf[ddf['value'] > 100] result = filtered.groupby('category').mean().compute() 要点：操作是惰性的（构建任务图），直到调用 .compute()；对高效自定义操作使用 map_partitions；在从其他来源处理结构化数据时，尽早转换为 DataFrame。
- Arrays：并行 NumPy 操作 用途：使用分块算法将 NumPy 能力扩展到大于内存的数据集。适用于数组超过可用 RAM、NumPy 操作需要并行化、处理科学数据集（HDF5、Zarr、NetCDF）、需要并行线性代数或数组操作等场景。 示例： import dask.array as da
创建带分块的大型数组
x = da.random.random((100000, 100000), chunks=(10000, 10000))
操作是惰性的
y = x + 100 z = y.mean(axis=0)
计算结果
result = z.compute() 要点：分块大小至关重要（目标每个分块约 100 MB）；操作在分块上并行执行；需要时重新分块以获得高效操作；对 Dask 中不可用的操作使用 map_blocks。
- Bags：非结构化数据的并行处理 用途：使用函数式操作处理非结构化或半结构化数据（文本、JSON、日志）。适用于处理文本文件、日志或 JSON 记录、在结构化分析前进行数据清洗与 ETL、处理不适合数组/DataFrame 格式的 Python 对象、需要内存高效的流式处理等场景。 示例： import dask.bag as db import json
读取并解析 JSON 文件
bag = db.read_text('logs/*.json').map(json.loads)
过滤并转换
valid = bag.filter(lambda x: x['status'] == 'valid') processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})
转换为 DataFrame 进行分析
ddf = processed.to_dataframe() 要点：用于初始数据清洗，然后转换为 DataFrame/Array；为获得更好性能，使用 foldby 而非 groupby；操作是流式且内存高效的；对复杂操作转换为结构化格式（DataFrame）。
- Futures：基于任务的并行化 用途：构建对任务执行与依赖有细粒度控制的自定义并行工作流。适用于构建动态、演进的工作流、需要立即执行任务（非惰性）、计算依赖运行时条件、实现自定义并行算法、需要有状态计算等场景。 示例： from dask.distributed import Client client = Client() # 创建本地集群
提交任务（立即执行）
def process(x): return x ** 2 futures = client.map(process, range(100))
收集结果
results = client.gather(futures) client.close() 要点：需要分布式客户端（即使单机也需要）；任务提交后立即执行；预先散布大型数据以避免重复传输；每个任务约 1ms 开销（不适合数百万个微小任务）；对有状态工作流使用 actors。
- Schedulers：执行后端 用途：控制 Dask 任务在何处及如何执行（线程、进程、分布式）。选择依据：Threads（默认）适合 NumPy/Pandas 操作、释放 GIL 的库、共享内存优势；Processes 适合纯 Python 代码、文本处理、受 GIL 限制的操作；Synchronous 适合用 pdb 调试、性能分析、理解错误；Distributed 适合需要仪表盘、多机集群和高级功能。 示例： import dask import dask.dataframe as dd
对 DataFrame 使用线程（默认，适合数值计算）
ddf = dd.read_csv('data.csv') result1 = ddf.mean().compute() # 使用线程
对 Python 密集型工作使用进程
import dask.bag as db bag = db.read_text('logs/*.txt') result2 = bag.map(python_function).compute(scheduler='processes')
使用同步调度器进行调试
dask.config.set(scheduler='synchronous') result3 = problematic_computation.compute() # 可使用 pdb
使用分布式进行监控和扩展
from dask.distributed import Client client = Client() result4 = computation.compute() # 使用带仪表盘的分布式 要点：线程开销最低（约 10 µs/任务），最适合数值工作；进程规避 GIL（约 10 ms/任务），最适合 Python 工作；分布式提供监控仪表盘（约 1 ms/任务），可扩展到集群；可按计算或全局切换调度器。
最佳实践 在引入 Dask 前先探索更简单的方案：更好的算法、高效文件格式（用 Parquet 替代 CSV）、编译代码（Numba、Cython）、数据采样。关键性能规则：不要先在本地加载数据再交给 Dask；避免重复调用 compute()；不要构建过大的任务图；选择合适的分块大小（目标每块约 100 MB）；使用仪表盘。
常见工作流模式 ETL 管道：用 dd.read_csv 读取多个原始文件，过滤、类型转换、去空值，再用 groupby 聚合并输出为 Parquet。 非结构化转结构化管道：用 Bag 读取并解析 JSON 日志，过滤后用 to_dataframe 转为 DataFrame 再做结构化分析。 大规模数组计算：用 da.from_zarr 加载大型数组，分块做归一化后写回 Zarr。 自定义并行工作流：用 Client 创建集群，scatter 大型数据集，submit 带依赖的并行任务，最后 gather 结果。
组件选择指南 按数据类型：表格数据 → DataFrames；数值数组 → Arrays；文本/JSON/日志 → Bags（再转为 DataFrame）；自定义 Python 对象 → Bags 或 Futures。按操作类型：标准 pandas 操作 → DataFrames；标准 NumPy 操作 → Arrays；自定义并行任务 → Futures；文本处理/ETL → Bags。按控制级别：高层自动 → DataFrames/Arrays；低层手动 → Futures。按工作流类型：静态计算图 → DataFrames/Arrays/Bags；动态演进 → Futures。
集成与格式 高效文件格式：Parquet、HDF5、Zarr（列式、压缩、对并行友好）；兼容但较慢：CSV（仅用于初始摄取）。数组：HDF5、Zarr、NetCDF。集合间转换：bag.to_dataframe()、ddf.to_dask_array(lengths=True)、dd.from_dask_array(arr, columns=['col1', 'col2'])。与其他库集成：XArray 为 Dask 数组包装带标签维度（地理空间、成像）；Dask-ML 提供与 scikit-learn 兼容的机器学习 API；Distributed 提供高级集群管理与监控。
调试与开发 迭代开发流程：先用同步调度器在小数据上测试，可用 pdb 调试；再用线程在样本上验证；最后用分布式监控扩展到全量数据。常见问题：内存错误可通过减小分块、策略性使用 persist() 并及时删除、检查自定义函数内存泄漏解决；启动慢常因任务图过大，可增大分块或用 map_partitions/map_blocks 减少任务；并行度差可因分块过大（增加分区）、Python 代码用线程（改用进程）或数据依赖阻碍并行。
参考文件 可根据需要阅读以下详细文档：references/dataframes.md（完整 Dask DataFrame 指南）、references/arrays.md（完整 Dask Array 指南）、references/bags.md（完整 Dask Bag 指南）、references/futures.md（Futures 与分布式计算指南）、references/schedulers.md（调度器选择与配置指南）、references/best-practices.md（性能优化与故障排除）。