dask 是做什麼的？
概觀
Dask 是一個用於平行與分散式運算的 Python 函式庫，提供三項關鍵能力：
- 大於記憶體的執行：在單機上處理超出可用 RAM 的資料
- 平行處理：透過多核心提升運算速度
- 分散式運算：跨多台機器支援 TB 級資料集
Dask 可從筆記型電腦（處理約 100 GiB）擴展到叢集（處理約 100 TiB），同時保持熟悉的 Python API。
當前上游版本： dask 2026.3.0（PyPI，2026 年 3 月）。文件：docs.dask.org。從 2025.1.0 起，基於表達式的 DataFrame API 搭配查詢計畫是唯一實作 — 請勿單獨安裝 dask-expr 或設定 dataframe.query-planning: False。
快速入門
安裝
uv pip install "dask>=2025.1"
若要搭配分散式排程器和儀表板使用典型的 Pandas/NumPy 工作流程：
uv pip install "dask[complete]"
遠端物件儲存（S3、GCS、Azure）：
uv pip install s3fs # s3:// 路徑
uv pip install gcsfs # gs:// 路徑
需要 Python 3.10+（2024.12 版已終止支援 3.9）。DataFrame I/O 需要 PyArrow 16+（自 dask 2026.1.2 起）。
何時使用此技能
此技能應在以下情況下使用：
- 處理超過可用 RAM 的資料集
- 將 Pandas 或 NumPy 操作擴展到更大的資料集
- 平行化運算以提升效能
- 有效率地處理多個檔案（CSV、Parquet、JSON、文字日誌）
- 建立具有任務依賴性的自訂平行工作流程
- 將工作負載分散到多個核心或機器上
核心功能
Dask 提供五個主要元件，每個都適用於不同的使用情境：
1. DataFrames - 平行 Pandas 操作
目的：透過平行處理將 Pandas 操作擴展到更大的資料集。
使用時機：
- 表格資料超過可用 RAM
- 需要一起處理多個 CSV/Parquet 檔案
- Pandas 操作速度慢且需要平行化
- 從 Pandas 原型擴展到生產環境
參考文件：有關 Dask DataFrames 的完整指南，請參閱 references/dataframes.md，其中包含：
- 讀取資料（單一檔案、多個檔案、全域模式）
- 常見操作（過濾、分組、合併、聚合）
- 使用
map_partitions的自訂操作 - 效能最佳化技巧
- 常見模式（ETL、時間序列、多檔案處理）
快速範例：
import dask.dataframe as dd
# 將多個檔案讀取為單一 DataFrame
ddf = dd.read_csv('data/2024-*.csv')
# 操作是延遲的，直到呼叫 compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()
重點：
- 操作是延遲的（建立任務圖），直到呼叫
.compute() - 使用
map_partitions進行高效的自訂操作 - 從其他來源處理結構化資料時，及早轉換為 DataFrame
2. Arrays - 平行 NumPy 操作
目的：使用分塊演算法將 NumPy 功能擴展到大於記憶體的資料集。
使用時機：
- 陣列超過可用 RAM
- NumPy 操作需要平行化
- 處理科學資料集（HDF5、Zarr、NetCDF）
- 需要平行線性代數或陣列操作
參考文件：有關 Dask Arrays 的完整指南，請參閱 references/arrays.md，其中包含：
- 建立陣列（從 NumPy、隨機、從磁碟）
- 分塊策略與最佳化
- 常見操作（算術、歸約、線性代數）
- 使用
map_blocks的自訂操作 - 與 HDF5、Zarr 和 XArray 的整合
快速範例：
import dask.array as da
# 建立具有分塊的大型陣列
x = da.random.random((100000, 100000), chunks=(10000, 10000))
# 操作是延遲的
y = x + 100
z = y.mean(axis=0)
# 計算結果
result = z.compute()
重點：
- 分塊大小至關重要（目標約 100 MB 每個分塊）
- 操作並行處理分塊
- 必要時重新分塊以進行高效操作
- 對 Dask 未提供的操作使用
map_blocks
3. Bags - 非結構化資料的平行處理
目的：使用函數式操作處理非結構化或半結構化資料（文字、JSON、日誌）。
使用時機：
- 處理文字檔、日誌或 JSON 記錄
- 在結構化分析之前進行資料清理和 ETL
- 處理不符合陣列/DataFrame 格式的 Python 物件
- 需要記憶體高效的串流處理
參考文件：有關 Dask Bags 的完整指南，請參閱 references/bags.md，其中包含：
- 讀取文字和 JSON 檔案
- 函數式操作（map、filter、fold、groupby）
- 轉換為 DataFrames
- 常見模式（日誌分析、JSON 處理、文字處理）
- 效能考量
快速範例：
import dask.bag as db
import json
# 讀取並解析 JSON 檔案
bag = db.read_text('logs/*.json').map(json.loads)
# 過濾和轉換
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})
# 轉換為 DataFrame 進行分析
ddf = processed.to_dataframe()
重點：
- 用於初始資料清理，然後轉換為 DataFrame/Array
- 使用
foldby代替groupby以獲得更好的效能 - 操作是串流式的且記憶體高效
- 對於複雜操作，轉換為結構化格式（DataFrame）
4. Futures - 基於任務的平行化
目的：建立具有精細控制任務執行和依賴性的自訂平行工作流程。
使用時機：
- 建立動態、不斷演變的工作流程
- 需要立即執行任務（非延遲）
- 計算依賴於執行階段的條件
- 實作自訂平行演算法
- 需要具有狀態的計算
參考文件：有關 Dask Futures 的完整指南，請參閱 references/futures.md，其中包含：
- 設定分散式客戶端
- 提交任務和使用 futures
- 任務依賴性和資料移動
- 進階協調（佇列、鎖、事件、執行者）
- 常見模式（參數掃描、動態任務、迭代演算法）
快速範例：
from dask.distributed import Client
client = Client() # 建立本地叢集
# 提交任務（立即執行）
def process(x):
return x ** 2
futures = client.map(process, range(100))
# 收集結果
results = client.gather(futures)
client.close()
重點：
- 需要分散式客戶端（即使是在單機上）
- 提交時任務立即執行
- 預先散布大型資料以避免重複傳輸
- 每個任務約 1ms 開銷（不適合數百萬個小型任務）
- 對具有狀態的工作流程使用執行者
5. 排程器 - 執行後端
目的：控制 Dask 任務在哪裡以及如何執行（執行緒、程序、分散式）。
何時選擇排程器：
- 執行緒（預設）：NumPy/Pandas 操作，釋放 GIL 的函式庫，共享記憶體效益
- 程序：純 Python 程式碼，文字處理，受 GIL 限制的操作
- 同步：使用 pdb 除錯、效能分析、了解錯誤
- 分散式：需要儀表板、多機器叢集、進階功能
參考文件：有關 Dask 排程器的完整指南，請參閱 references/schedulers.md，其中包含：
- 詳細的排程器描述和特性
- 設定方法（全域、上下文管理器、每次計算）
- 效能考量與開銷
- 常見模式和故障排除
- 最佳效能的執行緒設定
快速範例：
import dask
import dask.dataframe as dd
# 對 DataFrame 使用執行緒（預設，適合數值運算）
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute() # 使用執行緒
# 對 Python 密集型工作使用程序
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')
# 使用同步進行除錯
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute() # 可以使用 pdb
# 使用分散式進行監控和擴展
from dask.distributed import Client
client = Client()
result4 = computation.compute() # 使用分散式並搭配儀表板
重點：
- 執行緒：最低開銷（約 10 µs/任務），最適合數值運算
- 程序：避免 GIL（約 10 ms/任務），最適合 Python 工作
- 分散式：監控儀表板（約 1 ms/任務），可擴展到叢集
- 可以為每次計算或全域切換排程器
最佳實務
有關全面的效能最佳化指南、記憶體管理策略以及應避免的常見陷阱，請參閱 references/best-practices.md。關鍵原則包括：
從較簡單的方案開始
在使用 Dask 之前，請先探索：
- 更好的演算法
- 高效的檔案格式（Parquet 取代 CSV）
- 編譯程式碼（Numba、Cython）
- 資料抽樣
關鍵效能規則
1. 不要先在本地載入資料再交給 Dask
# 錯誤：先將所有資料載入記憶體
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)
# 正確：讓 Dask 處理載入
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')
2. 避免重複呼叫 compute()
# 錯誤：每次 compute 都是獨立的
for item in items:
result = dask_computation(item).compute()
# 正確：一次 compute 完成所有
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)
3. 不要建立過大的任務圖
- 如果有數百萬個任務，請增加分塊大小
- 使用
map_partitions/map_blocks來合併操作 - 檢查任務圖大小：
len(ddf.__dask_graph__())
4. 選擇適當的分塊大小
- 目標：每個分塊約 100 MB（或工作節點記憶體中每個核心 10 個分塊）
- 太大：記憶體溢位
- 太小：排程開銷
5. 使用儀表板
from dask.distributed import Client
client = Client()
print(client.dashboard_link) # 監控效能，找出瓶頸
常見工作流程模式
ETL 管道
import dask.dataframe as dd
# 提取：讀取資料
ddf = dd.read_csv('raw_data/*.csv')
# 轉換：清理和處理
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])
# 載入：聚合和儲存
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')
非結構化到結構化管道
import dask.bag as db
import json
# 使用 Bag 處理非結構化資料
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')
# 轉換為 DataFrame 進行結構化分析
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()
大規模陣列計算
import dask.array as da
# 載入或建立大型陣列
x = da.from_zarr('large_dataset.zarr')
# 分塊處理
normalized = (x - x.mean()) / x.std()
# 儲存結果（使用 mode= 覆寫；zarr_array_kwargs 用於壓縮）
da.to_zarr(normalized, 'normalized.zarr', mode='w')
自訂平行工作流程
from dask.distributed import Client
client = Client()
# 散布大型資料集一次
data = client.scatter(large_dataset)
# 平行處理並帶有依賴性
futures = []
for param in parameters:
future = client.submit(process, data, param)
futures.append(future)
# 收集結果
results = client.gather(futures)
選擇正確的元件
使用此決策指南來選擇適當的 Dask 元件：
資料類型：
- 表格資料 → DataFrames
- 數值陣列 → Arrays
- 文字/JSON/日誌 → Bags（然後轉換為 DataFrame）
- 自訂 Python 物件 → Bags 或 Futures
操作類型：
- 標準 Pandas 操作 → DataFrames
- 標準 NumPy 操作 → Arrays
- 自訂平行任務 → Futures
- 文字處理/ETL → Bags
控制程度：
- 高階、自動 → DataFrames/Arrays
- 低階、手動 → Futures
工作流程類型：
- 靜態計算圖 → DataFrames/Arrays/Bags
- 動態、不斷演變 → Futures
整合考量
檔案格式
- 高效能：Parquet、HDF5、Zarr（柱狀、壓縮、適合平行處理）
- 相容但較慢：CSV（僅用於初始匯入）
- 用於陣列：HDF5、Zarr、NetCDF
集合之間的轉換
# Bag → DataFrame
ddf = bag.to_dataframe()
# DataFrame → Array（用於數值資料）
arr = ddf.to_dask_array(lengths=True)
# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])
與其他函式庫的搭配
- XArray：使用帶標籤維度的 Dask 陣列（地理空間、影像）
- Dask-ML：使用與 scikit-learn 相容 API 的機器學習
- Distributed：進階叢集管理和監控
除錯與開發
迭代開發工作流程
- 在小型資料上使用同步排程器測試：
dask.config.set(scheduler='synchronous')
result = computation.compute() # 可以使用 pdb，易於除錯
- 在樣本上使用執行緒驗證：
sample = ddf.head(1000) # 小型樣本
# 測試邏輯，然後擴展到完整資料集
- 使用分散式進行監控擴展：
from dask.distributed import Client
client = Client()
print(client.dashboard_link) # 監控效能
result = computation.compute()
常見問題
記憶體錯誤：
- 減小分塊大小
- 策略性地使用
persist()，並在使用後刪除 - 檢查自訂函式中的記憶體洩漏
啟動緩慢：
- 任務圖太大（增加分塊大小）
- 使用
map_partitions或map_blocks減少任務數量
平行化不佳：
- 分塊太大（增加分割區數量）
- 使用執行緒處理 Python 程式碼（改為程序）
- 資料依賴性阻止平行化