dask

Dask - 可扩展并行计算

Dask 提供高级集合（数组、DataFrame、Bag），这些集合仿照 NumPy 和 pandas 的 API，但可并行处理大于内存容量的数据集。

适用场景

• 处理无法放入 RAM 的数据集（外存计算）。
• 利用所有可用的 CPU 核心加快计算。
• 并行执行自定义 Python 函数或复杂工作流（dask.delayed）。
• 将机器学习工作流扩展到大型集群。
• 处理物理学、气候科学或成像领域的大型数组。
• 分析海量日志文件或非结构化数据（dask.bag）。

参考文档

官方文档：https://docs.dask.org/
Dask 示例：https://examples.dask.org/
搜索关键词：dask.dataframe、dask.array、dask.delayed、client.compute、dask.distributed

核心原则

惰性求值

Dask 不会立即计算结果，而是构建任务图。只有显式调用 .compute() 或 .persist() 时，才会进行实际计算。

分块与分区

• Dask Array：由许多称为分块的小型 NumPy 数组组成。
• Dask DataFrame：由许多称为分区的小型 pandas DataFrame 组成。

适用 Dask 的场景

集合	类比	用例
dask.array	NumPy	大规模多维数学运算。
dask.dataframe	pandas	大型 CSV/Parquet/SQL 表。
dask.bag	Lists/Toolz	非结构化数据（JSON、日志）。
dask.delayed	Functions	自定义并行逻辑。

不适用 Dask 的场景

• 数据可以轻松放入 RAM（由于开销更低，pandas/NumPy 的速度更快）。
• 多进程或 concurrent.futures 足以处理的简单任务。
• 需要低延迟响应的场景（Dask 会增加调度开销）。

快速参考

安装

pip install "dask[complete]"

标准导入

import dask.array as da
import dask.dataframe as dd
from dask import delayed, compute
from dask.distributed import Client

基本模式 - 初始化本地集群

from dask.distributed import Client

# Setup local cluster and dashboard
client = Client() 
print(client.dashboard_link) # View real-time computation graph

关键规则

✅ 应当

• 使用仪表板 - 始终监控 Dask 仪表板以查找瓶颈（红色区块 = 不佳）。
• 合理分块 - 目标分块大小为 100MB 至 250MB。过小 = 开销高；过大 = 内存错误。
• 优先使用 Parquet - DataFrame 使用 Parquet 而不是 CSV；它支持高效的元数据和分区。
• 对重复使用的数据调用 .persist() - 如果多次使用同一个中间结果，将其持久化到内存中。
• 让 Dask 处理任务图 - 避免过早调用 .compute()；尽可能让计算保持惰性。
• 使用 map_partitions - 对 DataFrame 应用自定义逻辑时，使用它将 pandas 函数直接应用到每个分块。

❌ 不要

• 频繁计算 - 每次 .compute() 都会触发整个任务图的执行，并将数据载入 RAM。
• 向工作进程发送大型数据 - 对所有工作进程都需要的大型对象，使用 client.scatter，而不是将其作为参数传递。
• 逐行迭代 - for row in dask_df 会非常慢；请使用向量化操作。
• pandas 足够时仍使用 Dask - 对于小型数据，Dask 会因调度耗时而变慢。

反模式（绝不采用）

import dask.dataframe as dd

# ❌ BAD: Computing a large result into a local variable
# This will crash your local machine by filling RAM
result = dd_df.compute() 

# ✅ GOOD: Compute only what you need (aggregations)
mean_val = dd_df['column'].mean().compute()

# ❌ BAD: Too many small tasks (Task Overhead)
# result = [delayed(inc)(i) for i in range(1000000)] # 1 million tasks is too much

# ✅ GOOD: Batch tasks together or use Dask Collections
import dask.array as da
x = da.arange(1000000, chunks=10000) # Only 100 tasks

# ❌ BAD: Hardcoding workers' file paths
# dd.read_csv('/Users/me/data.csv') # Workers on other machines can't see this path!

# ✅ GOOD: Use shared storage (S3, HDFS, NFS)
# dd.read_csv('s3://my-bucket/data.csv')

Dask Array（dask.array）

扩展 NumPy

import dask.array as da

# Create a large random array (100GB)
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Perform operations (Lazy)
y = x + x.T
z = y[::2, :5000].mean(axis=0)

# Compute result
result = z.compute()

Dask DataFrame（dask.dataframe）

扩展 pandas

import dask.dataframe as dd

# Load massive dataset
df = dd.read_csv('data/*.csv')

# Filtering and Grouping
result = (df[df['value'] > 0]
          .groupby('category')
          .amount.sum())

# Execute
final_amounts = result.compute()

# Convert from pandas to dask
import pandas as pd
pdf = pd.DataFrame(...)
ddf = dd.from_pandas(pdf, npartitions=10)

Dask Delayed（dask.delayed）

并行执行自定义代码

from dask import delayed

@delayed
def load(filename):
    ...
    return data

@delayed
def process(data):
    ...
    return result

@delayed
def summarize(results):
    return sum(results)

# Build graph
filenames = ['file1.csv', 'file2.csv', 'file3.csv']
outputs = [process(load(f)) for f in filenames]
total = summarize(outputs)

# Visualize graph (requires graphviz)
# total.visualize()

# Execute in parallel
final_sum = total.compute()

使用 Dask 进行机器学习（dask-ml）

from dask_ml.preprocessing import StandardScaler
from dask_ml.linear_model import LogisticRegression
from dask_ml.model_selection import train_test_split

# Scaling to large data
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_dask)

# Training on large datasets (Parallel SGD)
model = LogisticRegression()
model.fit(X_dask, y_dask)

# Scikit-learn wrapper (for small data, parallelizing search)
from sklearn.ensemble import RandomForestClassifier
from dask_ml.model_selection import GridSearchCV

clf = RandomForestClassifier()
grid = GridSearchCV(clf, param_grid, cv=3)
grid.fit(X, y) # Grid search runs in parallel across cluster

实用工作流

1. 海量日志处理（Bag）

import dask.bag as db
import json

def analyze_logs(pattern):
    # Read unstructured text files
    b = db.read_text('logs/2023-*.log')
    
    # Parse JSON and filter
    records = b.map(json.loads).filter(lambda x: x['level'] == 'ERROR')
    
    # Extract specific field and count frequencies
    counts = records.pluck('message').frequencies()
    
    return counts.compute()

2. 大规模成像（Array）

def process_satellite_images(da_stack):
    """Calculate NDVI anomaly across time on 1TB of data."""
    # da_stack is a 3D dask array (time, x, y)
    
    # Simple vectorized math (Parallel)
    climatology = da_stack.mean(axis=0)
    anomaly = da_stack - climatology
    
    # Save results directly to disk without loading into RAM
    anomaly.to_zarr('anomalies.zarr')

3. 使用方法链清理数据

def clean_dataset(ddf):
    return (ddf
            .dropna(subset=['id'])
            .fillna({'status': 'unknown'})
            .assign(timestamp=dd.to_datetime(ddf['time_str']))
            .groupby('user_id')
            .last()
            .persist()) # Keep in memory for fast future use

性能优化

Dask 仪表板指南

• 进度条：显示已完成的任务数量。
• 任务流：显示每个工作进程正在执行的任务。空白区域 = 工作进程处于空闲状态（不佳）。
• 内存图：显示 RAM 使用情况。如果变为橙色/红色，说明工作进程已达到限制。
• 工作进程表：检查数据分布是否偏斜。

优化数据存储

• Zarr：最适合 N 维数组。
• Parquet：最适合表格型 DataFrame。
• 压缩：使用 snappy 或 lz4，在速度和大小之间取得平衡。

常见问题与解决方案

“Worker Lost” 错误

问题：Worker 因内存耗尽而崩溃。

解决方案：减小分块大小，或使用内存更大的机器。检查数据倾斜。

序列化错误（Pickle）

问题：Dask 无法将自定义对象发送给 Worker。

解决方案：使用 dask.distributed.Client.register_plugin，或确保类定义在 Worker 可以访问的单独文件中。

“Too Many Tasks” 警告

问题：你创建了 1,000,000 个以上的小任务。

解决方案：将数据重新分块为更大的部分。使用 dask_array.rechunk() 或 dask_df.repartition()。

最佳实践

1. 开发期间始终监控 Dask 仪表板，以识别瓶颈。
2. 仔细选择分块大小——为获得最佳性能，建议每块为 100-250MB。
3. 对于 DataFrame，使用 Parquet 格式而非 CSV，以获得更好的性能和元数据支持。
4. 对会重复使用的中间结果进行持久化，避免重新计算。
5. 尽可能延迟执行计算——仅在需要最终结果时调用 .compute()。
6. 对 Dask DataFrame 上的自定义 pandas 操作使用 map_partitions。
7. 避免遍历 Dask DataFrame 的行，改用向量化操作。
8. 在分布式集群上工作时使用共享存储（S3、HDFS、NFS）。
9. 将小任务批量处理，以避免任务开销。
10. 对能放入内存的数据不要使用 Dask——处理小型数据集时，pandas/NumPy 更快。

Dask 将 Python 从单线程脚本语言转变为世界一流的分布式计算系统。它是连接研究人员的笔记本电脑与高性能计算集群的桥梁。