Dask 是做什么的？
概述
Dask 是一个用于并行和分布式计算的 Python 库，提供三个关键能力：
- 超内存执行：在单机上处理超出可用 RAM 的数据
- 并行处理：跨多核提高计算速度
- 分布式计算：跨多台机器支持 TB 级数据集
Dask 可从笔记本电脑（处理约 100 GiB）扩展到集群（处理约 100 TiB），同时保持熟悉的 Python API。
当前上游版本： dask 2026.3.0（PyPI，2026 年 3 月）。文档：docs.dask.org。自 2025.1.0 起，基于表达式的 DataFrame API 和查询计划是唯一实现 — 不要单独安装 dask-expr 或设置 dataframe.query-planning: False。
快速开始
安装
uv pip install "dask>=2025.1"
对于典型的 pandas/NumPy 工作负载，包含分布式调度器和仪表板：
uv pip install "dask[complete]"
远程对象存储（S3、GCS、Azure）：
uv pip install s3fs # s3:// 路径
uv pip install gcsfs # gs:// 路径
需要 Python 3.10+（2024.12 版本已放弃 3.9 支持）。DataFrame I/O 需要 PyArrow 16+（自 dask 2026.1.2 起）。
何时使用此技能
当以下情况时应使用此技能：
- 处理超出可用 RAM 的数据集
- 将 pandas 或 NumPy 操作扩展到更大的数据集
- 并行化计算以提高性能
- 高效处理多个文件（CSV、Parquet、JSON、文本日志）
- 构建具有任务依赖关系的自定义并行工作流
- 将工作负载分布到多个核心或机器上
核心能力
Dask 提供五个主要组件，每个适用于不同的用例：
1. DataFrames - 并行 Pandas 操作
目的：通过并行处理将 pandas 操作扩展到更大的数据集。
何时使用：
- 表格数据超出可用 RAM
- 需要一起处理多个 CSV/Parquet 文件
- pandas 操作缓慢需要并行化
- 从 pandas 原型扩展到生产环境
参考文档：关于 Dask DataFrames 的全面指南，请参阅 references/dataframes.md，其中包括：
- 读取数据（单个文件、多个文件、通配符模式）
- 常见操作（过滤、分组、连接、聚合）
- 使用
map_partitions的自定义操作 - 性能优化技巧
- 常见模式（ETL、时间序列、多文件处理）
快速示例：
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
2. Arrays - 并行 NumPy 操作
目的：使用分块算法将 NumPy 能力扩展到超出内存的数据集。
何时使用：
- 数组超出可用 RAM
- NumPy 操作需要并行化
- 处理科学数据集（HDF5、Zarr、NetCDF）
- 需要并行线性代数或数组操作
参考文档：关于 Dask Arrays 的全面指南，请参阅 references/arrays.md，其中包括：
- 创建数组（从 NumPy、随机、从磁盘）
- 分块策略和优化
- 常见操作（算术、归约、线性代数）
- 使用
map_blocks的自定义操作 - 与 HDF5、Zarr 和 XArray 的集成
快速示例：
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
3. Bags - 非结构化数据的并行处理
目的：使用函数式操作处理非结构化或半结构化数据（文本、JSON、日志）。
何时使用：
- 处理文本文件、日志或 JSON 记录
- 在结构化分析之前进行数据清洗和 ETL
- 处理不适合数组/DataFrame 格式的 Python 对象
- 需要内存高效的流式处理
参考文档：关于 Dask Bags 的全面指南，请参阅 references/bags.md，其中包括：
- 读取文本和 JSON 文件
- 函数式操作（map、filter、fold、groupby）
- 转换为 DataFrames
- 常见模式（日志分析、JSON 处理、文本处理）
- 性能考虑
快速示例：
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）
4. Futures - 基于任务的并行化
目的：构建自定义并行工作流，对任务执行和依赖关系进行精细控制。
何时使用：
- 构建动态、演化的工作流
- 需要立即执行任务（非惰性）
- 计算依赖于运行时条件
- 实现自定义并行算法
- 需要带状态的计算
参考文档：关于 Dask Futures 的全面指南，请参阅 references/futures.md，其中包括：
- 设置分布式客户端
- 提交任务并操作 futures
- 任务依赖关系和数据移动
- 高级协调（队列、锁、事件、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()
关键点：
- 需要分布式客户端（即使是单机）
- 提交时任务立即执行
- 预先分散大型数据以避免重复传输
- 每个任务约 1ms 开销（不适合数百万个微小任务）
- 对有状态工作流使用 actors
5. Schedulers - 执行后端
目的：控制 Dask 任务的执行位置和方式（线程、进程、分布式）。
何时选择调度器：
- 线程（默认）：NumPy/Pandas 操作、释放 GIL 的库、共享内存有利
- 进程：纯 Python 代码、文本处理、GIL 绑定操作
- 同步：使用 pdb 调试、性能分析、理解错误
- 分布式：需要仪表板、多机集群、高级功能
参考文档：关于 Dask Schedulers 的全面指南，请参阅 references/schedulers.md，其中包括：
- 详细的调度器描述和特性
- 配置方法（全局、上下文管理器、每次计算）
- 性能考虑和开销
- 常见模式和故障排除
- 线程配置以获得最佳性能
快速示例：
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')
# 调试时使用同步
import dask.config as config
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），可扩展到集群
- 可以在每次计算或全局范围内切换调度器
最佳实践
有关全面的性能优化指南、内存管理策略和要避免的常见陷阱，请参阅 references/best-practices.md。关键原则包括：
从更简单的解决方案开始
在使用 Dask 之前，请探索：
- 更好的算法
- 高效的文件格式（Parquet 代替 CSV）
- 编译代码（Numba、Cython）
- 数据采样
关键性能规则
1. 不要先在本地加载数据再交给 Dask
# 错误：先将所有数据加载到内存中
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)
# 正确：让 Dask 处理加载
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')
2. 避免重复调用 compute()
# 错误：每次 compute 都是独立的
for item in items:
result = dask_computation(item).compute()
# 正确：一次 compute 完成所有计算
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)
3. 不要构建过大的任务图
- 如果有数百万个任务，则增加块大小
- 使用
map_partitions/map_blocks融合操作 - 检查任务图大小：
len(ddf.__dask_graph__())
4. 选择合适的块大小
- 目标：每个块约 100 MB（或每个核心在工作者内存中包含 10 个块）
- 太大：内存溢出
- 太小：调度开销
5. 使用仪表板
from dask.distributed import Client
client = Client()
print(client.dashboard_link) # 监控性能，识别瓶颈
常见工作流模式
ETL 流水线
import dask.dataframe as dd
# Extract：读取数据
ddf = dd.read_csv('raw_data/*.csv')
# Transform：清洗和处理
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])
# Load：聚合和保存
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')
非结构化到结构化流水线
import dask.bag as db
import json
# 从 Bag 开始处理非结构化数据
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')
# 转换为 DataFrame 进行结构化分析
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()
大规模数组计算
import dask.array as da
# 加载或创建大型数组
x = da.from_zarr('large_dataset.zarr')
# 按块处理
normalized = (x - x.mean()) / x.std()
# 保存结果（使用 mode= 覆盖；zarr_array_kwargs 用于压缩）
da.to_zarr(normalized, 'normalized.zarr', mode='w')
自定义并行工作流
from dask.distributed import Client
client = Client()
# 提前分散大型数据集一次
data = client.scatter(large_dataset)
# 带依赖关系的并行处理
futures = []
for param in parameters:
future = client.submit(process, data, param)
futures.append(future)
# 收集结果
results = client.gather(futures)
选择合适的组件
使用此决策指南来选择适当的 Dask 组件：
数据类型：
- 表格数据 → 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（仅用于初始导入）
- 用于 Arrays：HDF5、Zarr、NetCDF
集合之间的转换
# Bag → DataFrame
ddf = bag.to_dataframe()
# DataFrame → Array（用于数值数据）
arr = ddf.to_dask_array(lengths=True)
# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])
与其他库一起使用
- XArray：使用带标签维度的 Dask 数组封装（地理空间、成像）
- Dask-ML：具有 scikit-learn 兼容 API 的机器学习
- Distributed：高级集群管理和监控
调试与开发
迭代开发工作流
- 使用同步调度器在小数据上测试：
dask.config.set(scheduler='synchronous')
result = computation.compute() # 可以使用 pdb，易于调试
- 在样本上使用线程验证：
sample = ddf.head(1000) # 小样本
# 测试逻辑，然后扩展到完整数据集
- 使用分布式进行监控扩展：
from dask.distributed import Client
client = Client()
print(client.dashboard_link) # 监控性能
result = computation.compute()
常见问题
内存错误：
- 减小块大小
- 有策略地使用
persist()并在完成后删除 - 检查自定义函数中的内存泄漏
启动缓慢：
- 任务图太大（增加块大小）
- 使用
map_partitions或map_blocks减少任务数量
并行化不佳：
- 块太大（增加分区数）
- 对 Python 代码使用线程（切换到进程）
- 数据依赖项阻止并行执行
仪表板不显示：
- 确保未禁用仪表板
- 检查
client.dashboard_link是否正确 - 可能需要打开防火墙端口
参考文档
- dataframes.md - 有关 DataFrames 的全面指南
- arrays.md - 有关 Arrays 的全面指南
- bags.md - 有关 Bags 的全面指南
- futures.md - 有关 Futures 的全面指南
- schedulers.md - 有关 Schedulers 的全面指南
- best-practices.md - 最佳实践和陷阱
有关 Dask 生态系统的更多信息，请访问 docs.dask.org。