Dask 分布式计算专家

Dask

Dask 为 Python 提供扩展能力。它的 API 类似于 Pandas/NumPy，但可以在集群上运行。2025 年的更新重点在于高性能 Shuffle 和 GPU 集成。

何时使用

• 超内存表格计算：在单机或多节点集群上处理 10GB-1TB 的 CSV/Parquet 数据集。
• 并行化自定义 Python 工作流：使用 dask.delayed 并行化任意 Python 循环和任务图。
• 分布式机器学习：跨分布式工作节点与 Scikit-learn、XGBoost 和 LightGBM 集成。
• NumPy 与 Pandas 扩展：使用熟悉的 API 进行超内存数组和 DataFrame 计算。

快速开始

import dask.dataframe as dd

# 懒加载多 GB 分区 CSV 文件
df = dd.read_csv('data/transactions_*.csv')

# 跨所有 CPU 核心并行计算聚合统计量
result = df.groupby('category').amount.mean().compute()
print(result)

核心概念

#使用 Dask DataFrame 进行超内存处理

加载分区的 Parquet 文件并懒执行聚合计算：

import dask.dataframe as dd
from dask.distributed import Client

# 初始化本地分布式集群
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(f"Dask Dashboard 运行于: {client.dashboard_link}")

# 懒加载多文件数据集
df = dd.read_parquet(
    's3://analytics-bucket/transactions/*.parquet',
    columns=['customer_id', 'amount', 'status', 'country'],
    engine='pyarrow'
)

# 懒构建计算图
filtered = df[df['status'] == 'completed']
metrics = filtered.groupby('country')['amount'].agg(['mean', 'sum', 'count'])

# 执行计算图并返回具体的 Pandas DataFrame
result_df = metrics.compute()
print(result_df)

#使用 dask.delayed 自定义任务图

并行化独立函数执行：

from dask import delayed, compute
import time

@delayed
def fetch_api_data(endpoint: str) -> dict:
    time.sleep(0.5) # 模拟 I/O
    return {'endpoint': endpoint, 'status': 200, 'items': [1, 2, 3]}

@delayed
def transform_payload(data: dict) -> int:
    return sum(data['items'])

@delayed
def aggregate_totals(totals: list[int]) -> int:
    return sum(totals)

# 构建依赖 DAG
endpoints = ['users', 'orders', 'inventory', 'payments']
fetched = [fetch_api_data(ep) for ep in endpoints]
transformed = [transform_payload(f) for f in fetched]
final_total = aggregate_totals(transformed)

# 跨集群并行执行所有任务
total_result = final_total.compute()
print("总合:", total_result)

#Dask Array 用于分布式线性代数

操作大规模多维数组：

import dask.array as da

# 创建一个 50,000 x 50,000 的数组，分块大小为 5,000 x 5,000
x = da.random.normal(10, 0.1, size=(50000, 50000), chunks=(5000, 5000))

# 懒执行计算
y = x + x.T
z = y[::2, ::2].mean(axis=0)

# 计算结果
result = z.compute()
print("计算的数组均值形状:", result.shape)

常用模式

使用 Distributed Futures 处理不规则并行任务图

问题：不适合 DataFrame/Array 行列结构的自定义并行计算。

解决方案： 使用 Dask Distributed Client 配合异步 futures：

from dask.distributed import Client

client = Client(n_workers=4, threads_per_worker=2)

def train_partition(part_id: int):
    # 训练本地子模型
    return f"模型 {part_id} 已训练"

futures = [client.submit(train_partition, i) for i in range(10)]
results = client.gather(futures)
print(results)

最佳实践（2026）

• 应该 检查 Dask Web Dashboard（通常位于端口 8787）以监控内存压力、任务流和瓶颈。
• 应该 选择内存中 100MB 到 300MB 的分块大小以获得最佳并行效率。
• 应该 在重复查询相同转换后的数据时，持久化中间 DataFrame（df = df.persist()）。
• 应该 尽早使用列投影和谓词下推来过滤列和行。
• 不应该 在循环中调用 .compute()；应构建完整的任务图然后调用一次 compute()。
• 不应该 在数据完全适合内存时使用 Dask；原生 Pandas 和 Polars 在内存任务中显著更快。
• 不应该 创建数百万个小延迟任务；过多的任务开销会降低调度器性能。

故障排查

错误	原因	解决方案
KilledWorker: Worker exceeded 95% memory budget	计算时任务分区过大，超出工作节点内存。	增加分区数：df.repartition(npartitions=100)。
UserWarning: Sending large object to workers	将大型 NumPy 数组或 DataFrame 作为直接函数参数传递。	在提交任务前使用 client.scatter(data) 一次散布数据。
Slow compute: Too many tiny tasks	数千个微分区导致任务调度开销。	使用 repartition(partition_size="100MB") 合并小分区。

参考资料

• Dask 文档