findscripter Dask 超内存分布式计算
何时使用
适用场景：
- 数据超出可用内存（larger-than-RAM），需在单机做核外（out-of-core）计算（单机约可处理 ~100GiB，集群可达 ~100TiB）。
- 想把已有 pandas / NumPy 代码扩展到更大数据或跨多核/多机并行，沿用熟悉的 Python API。
- 批量处理多文件（CSV/Parquet/JSON/日志），或构建带任务依赖的自定义并行工作流。
不该用的边界：
- 数据能装进内存、只追求单机极速 → 用
polars-dataframe（表达式 + 并行更快）。 - 单机核外分析、不需要集群 → 可考虑 vaex。
- 先想更简单方案：换更好算法、用 Parquet 代替 CSV、Numba/Cython 编译、或数据采样，往往比上 Dask 划算。
- 不替代针对具体环境的校验、测试与专家评审；缺输入/权限/成功标准时先停下澄清。
步骤 / 指令
- 安装：
uv pip install "dask[complete]"（含分布式调度器与 dashboard）；远程对象存储另装s3fs（s3://）或gcsfs（gs://）。要求 Python 3.10+，DataFrame I/O 需 PyArrow 16+。 - 自 2025.1.0 起，表达式式 DataFrame API（带查询规划）是唯一实现——不要再单独装
dask-expr或设dataframe.query-planning: False。 - 按数据类型选组件：表格→DataFrame；数值数组→Array；文本/JSON/日志→Bag（清洗后转 DataFrame）；自定义动态任务→Futures。
- 让 Dask 自己读数据（如
dd.read_csv('data/2024-*.csv')），不要先 pandas 读全量再from_pandas。 - 操作是惰性的，构建任务图直到
.compute()才执行；多个结果用dask.compute(*xs)一次算完，别在循环里反复.compute()。 - 选合适分块/分区：目标 ~100MB/块（约每核 10 块）；过大爆内存、过小调度开销大；任务数百万时增大块或用
map_partitions/map_blocks融合操作。 - 选调度器：数值/释放 GIL 的库用 threads（默认，
10µs/task）；纯 Python 用 processes（10ms/task）；调试用 synchronous（可用 pdb）；要监控/集群用 distributed（Client()，~1ms/task）。 - 开 dashboard 定位瓶颈：
client = Client(); print(client.dashboard_link)。 - 落地用列式格式：
to_parquet/to_zarr（比 CSV 高效）。
迭代开发：先 dask.config.set(scheduler='synchronous') 小数据 + pdb 调试 → ddf.head(1000) 抽样验证逻辑 → 上 distributed 全量并监控。
示例
DataFrame（多文件 + 惰性聚合）：
import dask.dataframe as dd
ddf = dd.read_csv('data/2024-*.csv') # 多文件读成一个 DataFrame
filtered = ddf[ddf['value'] > 100] # 惰性
result = filtered.groupby('category').mean().compute()
ETL 管道并写 Parquet：
ddf = dd.read_csv('raw_data/*.csv')
ddf = ddf[ddf['status'] == 'valid']
ddf = ddf.dropna(subset=['important_col'])
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')
Array（分块 NumPy）：
import dask.array as da
x = da.random.random((100000, 100000), chunks=(10000, 10000))
z = (x + 100).mean(axis=0)
result = z.compute() # 块上并行；用 map_blocks 做缺失操作
Bag → DataFrame（非结构化清洗后转结构化）：
import dask.bag as db, json
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')
ddf = bag.to_dataframe() # 聚合用 foldby 优于 groupby
result = ddf.groupby('category').mean().compute()
Futures（自定义并行 + 预分发大数据）：
from dask.distributed import Client
client = Client()
data = client.scatter(large_dataset) # 大数据只 scatter 一次
futures = [client.submit(process, data, p) for p in parameters]
results = client.gather(futures) # 提交即执行，非惰性
注意事项
- 别先本地 pandas 读全量再交给 Dask——会先把全部数据塞内存，失去意义；让 Dask 直接
read_*。 - 别在循环里反复
.compute()；合并成dask.compute(*computations)一次算。 - 别建超大任务图：
len(ddf.__dask_graph__())查任务数，百万级就增大块或map_partitions/map_blocks融合。 - 分块大小是性能关键：~100MB/块；内存错误先减小块、酌情
persist()用完即删、排查自定义函数内存泄漏。 - 慢启动多因任务图太大；并行差常因块太大、threads 跑纯 Python（改 processes）或数据依赖阻塞并行。
- Futures 单任务约 1ms 开销，不适合百万级极小任务；有状态工作流用 actors，纯 Python 选 processes。
- 文件格式：Parquet/HDF5/Zarr（列式、压缩、并行友好）优先；CSV 仅用于初始摄入。
- 集合互转：
bag.to_dataframe()、ddf.to_dask_array(lengths=True)、dd.from_dask_array(arr, columns=[...])。
互见
- requires：无。
- related：
polars-dataframe（内存内极速，数据装得下时优先）；spark-job-optimization（更大规模/JVM 生态的分布式批处理）；scikit-learn-ml（配 Dask-ML 做分布式训练）。 - combines_with：
csv-data-cleaner（清洗后再上 Dask 扩展规模）；matplotlib-visualization/seaborn-statistical-charts（对.compute()结果或抽样做可视化）。
采编自 K-Dense-AI/scientific-agent-skills（MIT 许可）；该技能内容原署 K-Dense Inc.，原始声明为 BSD-3-Clause，基于 dask/dask，均可再分发。