dask-parallel-computing

Dask — 并行与分布式计算

概述

Dask 是一个用于并行与分布式计算的 Python 库，可将熟悉的 pandas/NumPy API 扩展到超出内存容量的数据集。它提供五个主要组件（DataFrames、Arrays、Bags、Futures、Schedulers），可从单机多核扩展到多节点 HPC 集群。

适用场景

• 处理超出可用 RAM 的数据集（10 GB–100 TB）
• 在多个核心上并行执行 pandas 或 NumPy 操作
• 高效处理多个文件（CSV、Parquet、JSON、HDF5、Zarr）
• 使用任务依赖关系构建自定义并行工作流
• 在 HPC 集群（SLURM、Kubernetes）中分发工作负载
• 处理非结构化数据（日志、JSON 记录）的流式/ETL 管道
• 对于单机内存内速度：改用 polars
• 对于单机外存分析：改用 vaex

前置条件

pip install dask[complete]        # All components
pip install dask[dataframe]       # DataFrames only
pip install dask[distributed]     # Distributed scheduler + dashboard
pip install dask-jobqueue          # HPC cluster integration (SLURM, PBS)

核心 API

1. DataFrames — 并行 pandas

import dask.dataframe as dd

# Read multiple files as a single DataFrame
ddf = dd.read_csv('data/2024-*.csv')
ddf = dd.read_parquet('data/', columns=['id', 'value', 'category'])

# Operations are lazy until .compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').agg({'value': ['mean', 'sum']}).compute()
print(result.shape)  # (n_categories, 2)

# Custom operations via map_partitions (preferred over apply)
def normalize_partition(df):
    df['norm_value'] = (df['value'] - df['value'].mean()) / df['value'].std()
    return df

ddf = ddf.map_partitions(normalize_partition)

# Joins
ddf_merged = ddf.merge(lookup_ddf, on='category', how='left')

# Write results
ddf.to_parquet('output/', engine='pyarrow')

# Repartitioning for optimal chunk sizes
ddf = ddf.repartition(npartitions=20)         # By count
ddf = ddf.repartition(partition_size='100MB')  # By size

# Index management for sorted operations
ddf = ddf.set_index('timestamp', sorted=True)

# Debugging
print(f"Partitions: {ddf.npartitions}")
print(f"Dtypes: {ddf.dtypes}")
sample = ddf.get_partition(0).compute()  # Inspect first partition

2. Arrays — 并行 NumPy

import dask.array as da
import numpy as np

# Create from various sources
x = da.random.random((100000, 1000), chunks=(10000, 1000))
x = da.from_array(np_array, chunks=(10000, 1000))
x = da.from_zarr('large_dataset.zarr')

# Standard operations (lazy)
y = (x - x.mean(axis=0)) / x.std(axis=0)  # Normalize
z = da.dot(x.T, x)                          # Matrix multiply
u, s, v = da.linalg.svd(x)                  # SVD

# Compute and persist
result = y.mean(axis=0).compute()
print(result.shape)  # (1000,)

# Custom operations with map_blocks
def custom_filter(block):
    from scipy.ndimage import gaussian_filter
    return gaussian_filter(block, sigma=2)

filtered = da.map_blocks(custom_filter, x, dtype=x.dtype)

# Rechunking for different access patterns
x_rechunked = x.rechunk({0: 5000, 1: 500})

# Save to disk
da.to_zarr(y, 'normalized.zarr')

3. Bags — 非结构化数据处理

import dask.bag as db
import json

# Read unstructured data
bag = db.read_text('logs/*.json').map(json.loads)

# Functional operations
valid = bag.filter(lambda x: x['status'] == 'success')
ids = valid.pluck('user_id')
flat = bag.map(lambda x: x['tags']).flatten()

# Aggregation — use foldby instead of groupby (much faster)
counts = bag.foldby(
    key='category',
    binop=lambda total, x: total + x['amount'],
    initial=0,
    combine=lambda a, b: a + b,
    combine_initial=0
).compute()

# Convert to DataFrame for structured analysis
ddf = valid.to_dataframe(meta={'user_id': 'str', 'amount': 'float64', 'category': 'str'})

4. Futures — 基于任务的并行处理

from dask.distributed import Client

client = Client()  # Local cluster with all cores
print(client.dashboard_link)  # http://localhost:8787

# Submit individual tasks (executes immediately, not lazy)
def process(x, param):
    return x ** param

future = client.submit(process, 42, param=2)
print(future.result())  # 1764

# Map over many inputs
futures = client.map(process, range(100), param=2)
results = client.gather(futures)
print(len(results))  # 100

# Scatter large data to workers (avoids repeated transfers)
import numpy as np
big_data = np.random.random((10000, 1000))
data_future = client.scatter(big_data, broadcast=True)

# Submit tasks using scattered data
futures = [client.submit(process_chunk, data_future, i) for i in range(10)]
results = client.gather(futures)

# Progressive result processing
from dask.distributed import as_completed
for future in as_completed(futures):
    result = future.result()
    print(f"Completed: {result}")

# Coordination primitives
from dask.distributed import Lock, Queue, Event
lock = Lock('resource-lock')
with lock:
    # Thread-safe operation across workers
    pass

client.close()

5. Schedulers 与配置

import dask

# Global scheduler setting
dask.config.set(scheduler='threads')       # Default: GIL-releasing numeric work
dask.config.set(scheduler='processes')     # Pure Python, GIL-bound work
dask.config.set(scheduler='synchronous')   # Debugging with pdb

# Context manager for temporary change
with dask.config.set(scheduler='synchronous'):
    result = computation.compute()  # Can use pdb here

# Per-compute override
result = ddf.mean().compute(scheduler='processes')

# Distributed scheduler with resource control
from dask.distributed import Client
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(client.dashboard_link)

# HPC cluster integration
from dask_jobqueue import SLURMCluster
from dask.distributed import Client

cluster = SLURMCluster(
    cores=24, memory='100GB',
    walltime='02:00:00', queue='regular'
)
cluster.scale(jobs=10)  # Request 10 SLURM jobs
client = Client(cluster)

# Adaptive scaling
cluster.adapt(minimum=2, maximum=20)

result = computation.compute()
client.close()

核心概念

组件选择指南

数据类型	组件	适用场景
表格数据（CSV、Parquet）	DataFrames	大规模执行类似 pandas 的标准操作
数值数组（HDF5、Zarr）	Arrays	NumPy 操作、线性代数、图像处理
文本、JSON、日志	Bags	ETL/清理 → 转换为 DataFrame 以供分析
自定义并行任务	Futures	动态工作流、参数扫描、任务依赖关系
上述任意类型	Schedulers	控制执行后端（线程/进程/分布式）

控制级别：DataFrames/Arrays/Bags = 高层惰性 API。Futures = 低层即时执行。

惰性求值模型

所有 DataFrames、Arrays 和 Bags 都会构建一个任务图——在调用 .compute() 或 .persist() 之前不会执行任何操作。

• .compute() — 执行并将结果返回到本地内存
• .persist() — 执行并将结果保留在工作进程上（以便在多个计算中重复使用）
• dask.compute(a, b, c) — 在一次遍历中计算多个结果（共享中间结果）

分块大小策略

目标：每个分块约 100 MB（或者工作进程内存中每个核心对应 10 个分块）。

分块大小	效果
过大（>1 GB）	内存溢出，并行化效果不佳
最佳（约 100 MB）	并行度良好，内存占用可控
过小（<1 MB）	调度开销过高

示例：8 个核心，32 GB RAM → 每个分块目标约 400 MB（32 GB / 8 个核心 / 10）。

调度器选择指南

调度器	开销	最适用于	GIL
threads（默认）	~10 µs/任务	NumPy、pandas、scikit-learn	受影响
processes	~10 ms/任务	纯 Python、文本处理	不受影响
synchronous	~1 µs/任务	使用 pdb 调试	不适用
distributed	~1 ms/任务	仪表板、集群、高级功能	可配置

常见工作流

工作流 1：多文件 ETL 管道

import dask.dataframe as dd
import dask

# Extract: Read all CSV files
ddf = dd.read_csv('raw_data/*.csv', dtype={'amount': 'float64'})

# Transform: Clean and process
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].fillna(0)
ddf = ddf.dropna(subset=['category'])

# Aggregate
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean', 'count']})

# Load: Save as Parquet (columnar, compressed)
summary.to_parquet('output/summary.parquet')
print(f"Processed {len(ddf)} rows across {ddf.npartitions} partitions")

工作流 2：大规模数组处理

import dask.array as da

# Load large scientific dataset
x = da.from_zarr('experiment_data.zarr')  # e.g., (50000, 50000) float64
print(f"Shape: {x.shape}, Chunks: {x.chunks}")

# Normalize per-column
x_norm = (x - x.mean(axis=0)) / x.std(axis=0)

# Compute covariance matrix
cov = da.dot(x_norm.T, x_norm) / (x_norm.shape[0] - 1)

# SVD for dimensionality reduction (top-k)
u, s, v = da.linalg.svd_compressed(x_norm, k=50)

# Save results
da.to_zarr(u, 'pca_components.zarr')
print(f"Explained variance (top 5): {(s[:5]**2 / (s**2).sum()).compute()}")

工作流 3：将非结构化数据用于分析

此工作流将 Bags（第 3 节）与 DataFrames（第 1 节）简单组合：使用 Bags 读取 JSON 日志，进行筛选/转换，再转换为 DataFrame 以进行 groupby 分析。每个步骤都直接对应上文核心 API 中的示例。

关键参数

参数	模块	默认值	描述
npartitions	DataFrame	auto	分区数量（控制并行度）
partition_size	DataFrame	—	每个分区的目标大小（例如，'100MB'）
chunks	Array	required	分块维度（例如，(10000, 1000)）
blocksize	Bag	'128 MiB'	文件读取块大小
scheduler	All	'threads'	执行后端（'threads'、'processes'、'synchronous'）
n_workers	Distributed	auto	工作进程数量
threads_per_worker	Distributed	auto	每个工作进程的线程数
memory_limit	Distributed	auto	每个工作进程的内存限制（例如，'4GB'）
sorted	set_index	False	数据是否已预先排序（可启用优化）
meta	map_partitions	—	输出 DataFrame/Series 结构模板

最佳实践

1. 让 Dask 负责加载数据 — 切勿先将数据加载到 pandas/numpy 中再进行转换。直接使用 dd.read_csv() / da.from_zarr()。

2. 批量调用计算 — 使用 dask.compute(a, b, c)，而不是在循环中调用 .compute()。这样可以共享中间结果。

3. 优先使用 map_partitions，而非 apply — ddf.apply(func, axis=1) 会为每一行创建一个任务。ddf.map_partitions(func) 则会为每个分区创建一个任务。

4. 持久化重复使用的中间结果 — 对多次访问的数据调用 .persist()，完成后再调用 del。

5. 使用仪表板 — client.dashboard_link 可显示任务进度、内存用量和工作进程状态，是排查性能问题的必备工具。

6. 反模式 — 任务图过大：如果 len(ddf.__dask_graph__()) 返回数百万，增大分块大小，或使用 map_partitions/map_blocks 合并操作。

7. 反模式 — 为工作负载选择了错误的调度器：对纯 Python 文本处理（受 GIL 限制）使用线程，或对 NumPy 操作使用进程（会产生不必要的序列化开销）。

常用示例

示例：Dask-ML 集成

from dask_ml.preprocessing import StandardScaler
from dask_ml.model_selection import train_test_split
import dask.array as da

X = da.random.random((100000, 50), chunks=(10000, 50))
y = da.random.randint(0, 2, size=100000, chunks=10000)

# Preprocessing
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# Train/test split
X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2)
print(f"Train: {X_train.shape}, Test: {X_test.shape}")

示例：用于有状态计算的 Actors

from dask.distributed import Client

client = Client()

class RunningStats:
    def __init__(self):
        self.count = 0
        self.total = 0.0

    def add(self, value):
        self.count += 1
        self.total += value
        return self.total / self.count

# Create actor on a worker
stats = client.submit(RunningStats, actor=True).result()

# Call methods (~1ms roundtrip)
for v in [10, 20, 30]:
    mean = stats.add(v).result()
    print(f"Running mean: {mean}")

client.close()

示例：Kubernetes 集群

from dask_kubernetes import KubeCluster
from dask.distributed import Client

cluster = KubeCluster()
cluster.adapt(minimum=2, maximum=50)
client = Client(cluster)

# Run computation on auto-scaling cluster
result = large_computation.compute()
client.close()

故障排除

问题	原因	解决方案
.compute() 期间出现 MemoryError	结果对本地内存来说过大	使用 .to_parquet() 或 .persist()，而非 .compute()
计算启动缓慢	任务图包含数百万个任务	增大分块大小；使用 map_partitions/map_blocks
并行化效果差	使用线程调度器时发生 GIL 争用	对 Python 密集型代码切换为 scheduler='processes'
map_partitions 中出现 TypeError	缺少或错误的 meta 参数	提供 meta=pd.DataFrame({'col': pd.Series(dtype='float64')})
工作进程被终止（OOM）	分块超出工作进程的内存容量	减小分块大小；增大 memory_limit
KilledWorker 异常	工作进程崩溃	检查工作进程日志；减少每个任务的内存用量；增大 memory_limit
联接/合并缓慢	数据未按联接键预先排序	在联接前调用 ddf.set_index('key', sorted=True)
NotImplementedError	Dask 不支持该操作	使用 map_partitions 搭配 pandas 中的对应操作
无法访问仪表板	尚未启动 Distributed 客户端	使用 client = Client() 启用分布式调度器
各分区之间的数据类型不一致	CSV 文件内容不一致	在 dd.read_csv() 中明确指定 dtype

打包的资源

• references/collections_guide.md — 详细介绍 DataFrames、Arrays 和 Bags 的指南，包含丰富的代码示例，涵盖数据读取、转换、聚合和写入。介绍 map_partitions 模式、meta 参数、分块策略、map_blocks、foldby 和集合转换。由原来的 dataframes.md、arrays.md 和 bags.md 整合而成。原 best-practices.md 的内容已移至最佳实践部分，并以内联形式纳入关键概念。
• references/distributed_computing.md — Futures API、分布式协调原语（Locks、Queues、Events、Variables）、Actors、调度器配置、HPC 集群设置（SLURM、Kubernetes）、自适应扩缩、仪表板监控和性能分析。由原来的 futures.md 和 schedulers.md 整合而成。

未迁移：原来有 6 个参考文件。best-practices.md 的内容已整合到最佳实践部分和关键概念中（分块策略、调度器选择）。其余内容整理到 2 个参考文件中，涵盖 5 个主要组件。

相关技能

• polars-dataframes — 用于单机内存处理的 DataFrame 库；对于可放入 RAM 的数据，速度比 Dask 更快
• zarr-python — 分块数组存储格式；Dask Array 的主要持久化后端
• scikit-learn-machine-learning — 机器学习库；Dask-ML 提供分布式封装器

参考资料

• 官方文档：https://docs.dask.org/
• Dask 最佳实践：https://docs.dask.org/en/stable/best-practices.html
• Dask-ML：https://ml.dask.org/
• Dask-Jobqueue（HPC）：https://jobqueue.dask.org/