dask

Dask

概述

Dask 是一个用于并行和分布式计算的 Python 库，提供三项关键能力：

• 在单台机器上处理超出内存容量的数据，以运行超出可用 RAM 的数据集
• 并行处理，利用多个核心提升计算速度
• 分布式计算，支持在多台机器上处理太字节级的数据集

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:// paths
uv pip install gcsfs   # gs:// paths

需要 Python 3.10+（已于 2024.12 停止支持 3.9）。DataFrame I/O 需要 PyArrow 16+（截至 dask 2026.1.2）。

何时使用此技能

以下情况应使用此技能：

• 处理超出可用 RAM 容量的数据集
• 将 pandas 或 NumPy 操作扩展到更大的数据集
• 并行化计算以提升性能
• 高效处理多个文件（CSV、Parquet、JSON、文本日志）
• 构建带有任务依赖关系的自定义并行工作流
• 在多个核心或机器之间分发工作负载

核心能力

详细信息、代码示例和参数表：references/core-capabilities.md。在适用此步骤时阅读。

最佳实践

如需全面了解性能优化建议、内存管理策略和应避免的常见问题，请参阅 references/best-practices.md。主要原则包括：

从更简单的方案入手

使用 Dask 前，请先考虑：

• 更好的算法
• 高效的文件格式（使用 Parquet 而不是 CSV）
• 编译代码（Numba、Cython）
• 数据采样

关键性能规则

1. 不要先在本地加载数据，再交给 Dask

# Wrong: Loads all data in memory first
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correct: Let Dask handle loading
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. 避免重复调用 compute()

# Wrong: Each compute is separate
for item in items:
    result = dask_computation(item).compute()

# Correct: Single compute for all
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)  # Monitor performance, identify bottlenecks

常见工作流模式

ETL 流水线

import dask.dataframe as dd

# Extract: Read data
ddf = dd.read_csv('raw_data/*.csv')

# Transform: Clean and process
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load: Aggregate and save
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

从非结构化数据到结构化数据的流水线

import dask.bag as db
import json

# Start with Bag for unstructured data
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Convert to DataFrame for structured analysis
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

大规模数组计算

import dask.array as da

# Load or create large array
x = da.from_zarr('large_dataset.zarr')

# Process in chunks
normalized = (x - x.mean()) / x.std()

# Save result (use mode= for overwrite; zarr_array_kwargs for compression)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

自定义并行工作流

from dask.distributed import Client

client = Client()

# Scatter large dataset once
data = client.scatter(large_dataset)

# Process in parallel with dependencies
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Gather results
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（仅用于初始导入）
• 用于数组：HDF5、Zarr、NetCDF

集合之间的转换

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (for numeric data)
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：高级集群管理和监控

调试与开发

迭代式开发工作流

1. 使用同步调度器在小规模数据上测试：

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Can use pdb, easy debugging

2. 使用线程在样本上验证：

sample = ddf.head(1000)  # Small sample
# Test logic, then scale to full dataset

3. 使用分布式模式扩展并进行监控：

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitor performance
result = computation.compute()

常见问题

内存错误：

• 减小数据块大小
• 有策略地使用 persist()，并在完成后删除
• 检查自定义函数中是否存在内存泄漏

启动缓慢：

• 任务图过大（增大数据块大小）
• 使用 map_partitions 或 map_blocks 减少任务数

并行化效果不佳：

• 数据块过大（增加分区数）
• 使用线程运行 Python 代码（改用进程）
• 数据依赖阻碍并行处理

参考文件

可根据需要读取所有参考文档文件，以获取详细信息：

• references/dataframes.md - 完整的 Dask DataFrame 指南
• references/arrays.md - 完整的 Dask Array 指南
• references/bags.md - 完整的 Dask Bag 指南
• references/futures.md - 完整的 Dask Futures 和分布式计算指南
• references/schedulers.md - 完整的调度器选择与配置指南
• references/best-practices.md - 全面的性能优化与故障排查指南

当用户需要了解此处快速指南未涵盖的特定 Dask 组件、操作或模式的详细信息时，请加载这些文件。

Agent 操作流程

1. 检查环境。 确认 Python 环境和库版本（python -c "import pkg; print(pkg.__version__)"），并检查数据的形状、类型和缺失值。
2. 明确输入。 根据数据或用户确认格式、标识符和参数。对于任何会影响结果的值，应先询问，不要猜测。
3. 先运行小规模版本。 先使用样本或单个折运行，并检查运行时间和内存。
4. 执行完整任务，遵循上述说明和参考资料。
5. 验证结果。 使用留出数据、固定随机种子和适当的指标；检查是否存在数据泄漏；报告不确定性（置信区间、不同随机种子的标准差）。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及哪些方面仍不确定。

如果发生这种情况	请这样做
内存不足或执行速度极慢	进行子采样，使用分块或惰性计算，或缩小模型规模，并告知用户所做的更改。
这些说明中的函数、标志或端点在已安装版本中不存在	查阅已安装版本自己的文档（help()、--help、官方文档），进行调整并告知用户。切勿虚构 API。
必需的输入、标识符或参数不明确	询问用户，或在运行前明确说明假设。

诚信守则

• 切勿捏造结果、参数、标识符、引用或统计数据。如果无法运行或验证某项内容，请明确说明。
• 切勿报告本次会话中未计算的指标；展示生成每个数字的代码路径。
• 此处与版本相关的细节可能已经过时：请对照已安装版本的官方文档进行确认。
• 对于会产生费用、消耗共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，执行前必须征得同意。

相关技能

• polars：用于 Python ETL、分析和 pandas 迁移的高性能 DataFrame 库。
• ray-data：面向 ML 工作负载的可扩展数据处理。
• ray-train：跨集群的分布式训练编排。