dask-parallel-computing

Dask — 並列・分散コンピューティング

概要

Dask は、使い慣れた pandas／NumPy API をメモリ容量を超えるデータセットにも拡張できる、並列・分散コンピューティング向けの Python ライブラリです。5 つの主要コンポーネント（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 は低水準の即時実行 API です。

遅延評価モデル

すべての DataFrames、Arrays、Bags はタスクグラフを構築し、.compute() または .persist() が呼び出されるまで実行されません。

• .compute() — 実行し、結果をローカルメモリに返します
• .persist() — 実行し、結果をワーカー上に保持します（複数の計算で再利用する場合）
• dask.compute(a, b, c) — 1 回の処理で複数の結果を計算します（中間結果を共有します）

チャンクサイズの方針

目標：チャンクあたり約 100 MB（またはワーカーのメモリ内でコアあたり 10 チャンク）。

チャンクサイズ	影響
大きすぎる（>1 GB）	メモリ不足、並列化の効率低下
最適（約 100 MB）	優れた並列性、管理しやすいメモリ使用量
小さすぎる（<1 MB）	スケジューリングのオーバーヘッドが過大

例：8 コア、32 GB RAM の場合 → チャンクあたり約 400 MB を目標とします（32 GB / 8 コア / 10）。

Scheduler の選択ガイド

Scheduler	オーバーヘッド	最適な用途	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 ログを読み込み、フィルター処理と変換を行い、groupby 分析のために DataFrame へ変換します。各ステップは、上記のコア 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. compute の呼び出しをまとめる — ループ内で .compute() を呼び出す代わりに、dask.compute(a, b, c) を使用します。中間結果を共有できます。

3. apply より map_partitions を使用する — ddf.apply(func, axis=1) は行ごとに 1 つのタスクを作成します。ddf.map_partitions(func) はパーティションごとに 1 つのタスクを作成します。

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}")

レシピ：状態を保持する計算でのアクター

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	結果がローカルメモリーに収まらない	.compute() の代わりに .to_parquet() または .persist() を使用します
計算開始が遅い	タスクグラフに数百万のタスクがある	チャンクサイズを大きくし、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 でサポートされていない操作	pandas 相当の処理を使って map_partitions を使用します
ダッシュボードにアクセスできない	Distributed クライアントが起動していない	client = Client() を使用して分散スケジューラーを有効にします
パーティション間でデータ型が一致しない	CSV ファイル間で内容に一貫性がない	dd.read_csv() で dtype を明示的に指定します

同梱リソース

• references/collections_guide.md — DataFrames、Arrays、Bag の詳細ガイドです。データの読み込み、変換、集計、書き込みに関する包括的なコード例を掲載しています。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 の内容は「ベストプラクティス」セクションと「主要な概念」（チャンク化戦略、スケジューラーの選択）に統合しました。残りの内容は、5つの主要コンポーネントを扱う2つのリファレンスファイルに整理しました。

関連スキル

• 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/