Dask 是 Python 的灵活并行计算库，可将 NumPy、pandas、scikit-learn 等 Python 库的工作负载扩展到多核系统或分布式集群。它提供高层集合（Arrays、DataFrames、Bags），模仿 NumPy 与 pandas 的 API，但能并行处理大于内存的数据集。
【用途与适用场景】 处理无法装入 RAM 的数据集（out-of-core 计算）；利用全部 CPU 核心加速计算；并行化自定义 Python 函数或复杂工作流（dask.delayed）；将机器学习流水线扩展到大型集群；处理物理、气候科学或成像领域的大规模数组；分析海量日志文件或非结构化数据（dask.bag）。
【核心原理】
- 延迟求值（Lazy Evaluation）：Dask 不立即计算结果，而是构建任务图（Task Graph），只有显式调用 .compute() 或 .persist() 时才真正计算。
- 分块与分区：Dask Array 由许多小 NumPy 数组（chunks）组成；Dask DataFrame 由许多小 pandas DataFrame（partitions）组成。
【集合对应关系】 dask.array 对应 NumPy，用于大规模多维数学计算；dask.dataframe 对应 pandas，用于大型 CSV/Parquet/SQL 表；dask.bag 对应列表/Toolz，用于非结构化数据（JSON、日志）；dask.delayed 对应函数，用于自定义并行逻辑。
【安装与导入】 安装命令：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 client = Client() print(client.dashboard_link) # 查看实时计算图
【关键规则】 建议：使用 Dashboard 监控瓶颈（红色块表示有问题）；合理设置分块大小，以 100MB 到 250MB 为目标；DataFrame 优先使用 Parquet 而非 CSV；对重复使用的数据调用 .persist() 保留在内存；尽量保持计算惰性，避免过早调用 .compute()；对 DataFrame 自定义逻辑使用 map_partitions。 避免：过于频繁计算；向 worker 传递大数据（应使用 client.scatter）；逐行迭代 DataFrame（极慢，应用向量化操作）；在 pandas 足够时使用 Dask（小数据下 Dask 因调度开销反而更慢）。
【Dask Array 示例】 import dask.array as da x = da.random.random((100000, 100000), chunks=(10000, 10000)) y = x + x.T z = y[::2, :5000].mean(axis=0) result = z.compute()
【Dask DataFrame 示例】 import dask.dataframe as dd df = dd.read_csv('data/*.csv') result = (df[df['value'] > 0].groupby('category').amount.sum()) final_amounts = result.compute() 从 pandas 转换：ddf = dd.from_pandas(pdf, npartitions=10)
【Dask Delayed 示例】 from dask import delayed @delayed def load(filename): ... @delayed def process(data): ... @delayed def summarize(results): return sum(results) filenames = ['file1.csv', 'file2.csv', 'file3.csv'] outputs = [process(load(f)) for f in filenames] total = summarize(outputs) final_sum = total.compute()
【机器学习（dask-ml）】 from dask_ml.preprocessing import StandardScaler from dask_ml.linear_model import LogisticRegression scaler = StandardScaler() X_scaled = scaler.fit_transform(X_dask) model = LogisticRegression() model.fit(X_dask, y_dask) 亦可用 dask_ml.model_selection.GridSearchCV 在集群上并行执行网格搜索。
【实践工作流】 日志处理（Bag）：db.read_text 读取日志，map(json.loads) 解析、filter 过滤，pluck 提取字段并 frequencies 统计。 大规模成像（Array）：对三维 dask 数组求均值得到气候态、计算异常，并用 to_zarr 直接写盘，无需载入 RAM。 数据清洗：方法链式调用 dropna、fillna、assign、groupby().last()，最后 persist()。
【性能优化与 Dashboard】 Dashboard 提供进度条、Task Stream（空白表示 worker 空闲）、Memory Plot（橙红表示接近内存上限）、Worker Table（检查数据倾斜）。存储方面：Zarr 适合 N 维数组，Parquet 适合表格 DataFrame，压缩可选 snappy 或 lz4。
【常见问题】 Worker Lost：worker 内存耗尽崩溃，应减小 chunk 或增加内存，并排查数据倾斜。序列化（Pickle）错误：Dask 无法将自定义对象发送给 worker，可使用 Client.register_plugin 或确保类定义在 worker 可访问的独立文件中。任务过多警告：创建了百万级小任务，应使用 rechunk() 或 repartition() 重新分块。
【限制】 数据能轻松装入 RAM 时不建议使用 Dask（pandas/NumPy 因开销更低而更快）；简单任务用 multiprocessing 或 concurrent.futures 即可；需要低延迟响应时不适合（Dask 有调度开销）；引用文档与示例分别见 https://docs.dask.org/ 与 https://examples.dask.org/。