dask

用途

Dask 是一个用于并行与分布式计算的 Python 库，旨在让现有的 pandas/NumPy 代码能够扩展到超出内存容量或跨集群运行。它提供三项关键能力：单机上的超内存执行（处理超过可用 RAM 的数据）、多核并行处理以提升计算速度，以及支持跨多机处理 TB 级数据集的分布式计算。Dask 可从笔记本（约 100 GiB）扩展到集群（约 100 TiB），同时保持熟悉的 Python API。

适用时机

当需要处理超出可用 RAM 的数据集、将 pandas 或 NumPy 操作扩展到更大数据、并行化计算以提升性能、高效处理多个文件（CSV、Parquet、JSON、文本日志）、构建带任务依赖的自定义并行工作流，或将工作负载分发到多核或多机时，可使用本技能。

主要能力

Dask 提供五个主要组件，各自适用于不同场景：

1. DataFrames — 并行 Pandas 操作 用于通过并行处理将 pandas 操作扩展到更大数据集。适用于表格数据超出 RAM、需要一起处理多个 CSV/Parquet 文件、pandas 操作缓慢需要并行化，或从 pandas 原型扩展到生产环境。操作是惰性的（构建任务图），直到调用 .compute() 才执行；可使用 map_partitions 进行高效自定义操作。示例：

import dask.dataframe as dd

ddf = dd.read_csv('data/2024-*.csv')
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

2. Arrays — 并行 NumPy 操作 用于通过分块算法将 NumPy 能力扩展到超出内存的数据集。适用于数组超出 RAM、NumPy 操作需要并行化、处理科学数据集（HDF5、Zarr、NetCDF），或需要并行线性代数/数组操作。分块大小至关重要（目标约 100 MB 每块）；可使用 map_blocks 处理 Dask 中不可用的操作。示例：

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()

3. Bags — 非结构化数据的并行处理 用于以函数式操作处理非结构化或半结构化数据（文本、JSON、日志）。适用于处理文本文件、日志或 JSON 记录，在结构化分析前进行数据清洗和 ETL，处理不适合数组/数据框格式的 Python 对象，或需要内存高效的流式处理。建议用于初始数据清洗，然后转换为 DataFrame/Array；使用 foldby 而非 groupby 以获得更好性能。示例：

import dask.bag as db
import 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']})
ddf = processed.to_dataframe()

4. Futures — 基于任务的并行化 用于构建对任务执行和依赖有细粒度控制的自定义并行工作流。适用于构建动态、演进的工作流，需要立即执行任务（非惰性），计算依赖运行时条件，实现自定义并行算法，或需要有状态计算。需要分布式客户端（即使单机）；任务提交后立即执行；建议预先 scatter 大数据以避免重复传输；每个任务约 1ms 开销（不适合数百万微小任务）；可使用 actors 实现有状态工作流。示例：

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()

5. Schedulers — 执行后端 用于控制 Dask 任务的执行方式和位置（线程、进程、分布式）。选择建议：线程（默认）适合 NumPy/Pandas 操作、释放 GIL 的库、共享内存；进程适合纯 Python 代码、文本处理、受 GIL 限制的操作；同步适合用 pdb 调试、性能分析、理解错误；分布式适合需要仪表盘、多机集群和高级功能。可全局或按计算切换调度器。示例：

import dask
import dask.dataframe as dd

ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # 使用线程

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()

from dask.distributed import Client
client = Client()
result4 = computation.compute()

使用方式与最佳实践

在采用 Dask 前，应先探索更简单的方案：更好的算法、高效文件格式（Parquet 而非 CSV）、编译代码（Numba、Cython）、数据采样。关键性能规则包括：不要让 Dask 先本地加载数据再交给它（应使用 dd.read_csv 直接读取）；避免重复调用 compute()（应使用 dask.compute(*computations) 一次性计算）；不要构建过大的任务图（增大分块、使用 map_partitions/map_blocks 融合操作、检查 len(ddf.__dask_graph__())）；选择合适的分块大小（目标约 100 MB 每块）；使用仪表盘监控性能（client.dashboard_link）。

常见工作流模式包括 ETL 管道、非结构化到结构化管道、大规模数组计算和自定义并行工作流。选择组件时可参考：表格数据用 DataFrames，数值数组用 Arrays，文本/JSON/日志用 Bags（再转 DataFrame），自定义 Python 对象用 Bags 或 Futures；标准 pandas/NumPy 操作用 DataFrames/Arrays，自定义并行任务用 Futures，文本处理/ETL 用 Bags；高层自动用 DataFrames/Arrays，底层手动用 Futures；静态计算图用 DataFrames/Arrays/Bags，动态演进用 Futures。

集成方面：高效文件格式为 Parquet、HDF5、Zarr；数组可用 HDF5、Zarr、NetCDF；集合间可转换（bag.to_dataframe()、ddf.to_dask_array(lengths=True)、dd.from_dask_array(arr, columns=[...])）；可与 XArray、Dask-ML、Distributed 配合使用。

调试与开发建议采用迭代工作流：先用同步调度器在小数据上测试，再用线程在样本上验证，最后用分布式扩展并监控。常见问题包括内存错误（减小分块、策略性使用 persist() 并及时删除、检查自定义函数内存泄漏）、启动缓慢（任务图过大，增大分块、使用 map_partitions/map_blocks 减少任务）、并行化不佳（分块过大、Python 代码使用线程、数据依赖阻碍并行）。

限制

资料中未提供关于收费、价格、平台支持范围或具体性能保证的明确信息。Dask 的 Futures 每个任务约有 1ms 开销，不适合数百万微小任务；分块过大可能导致内存溢出，过小则增加调度开销。