dask

Dask - スケーラブルな並列計算

Dask は、NumPy や pandas の API を模倣しながら、メモリに収まらない大きさのデータセットを並列処理する、高水準のコレクション（Arrays、DataFrames、Bags）を提供します。

使用する場面

• 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	リスト／Toolz	非構造化データ（JSON、ログ）。
dask.delayed	関数	独自の並列ロジック。

使用しない場面

• RAM に余裕をもって収まるデータ（オーバーヘッドが小さいため、pandas／NumPy のほうが高速です）。
• multiprocessing または 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 では CSV ではなく Parquet を使います。効率的なメタデータとパーティショニングに対応しています。
• 再利用するデータには .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」エラー

問題：ワーカーの RAM が不足し、クラッシュします。

解決策：チャンクサイズを小さくするか、より多くのメモリを搭載したマシンを使用します。データの偏りがないか確認してください。

シリアライズエラー（Pickle）

問題：Dask がカスタムオブジェクトをワーカーに送信できません。

解決策：dask.distributed.Client.register_plugin を使用するか、ワーカーからアクセスできる別ファイルでクラスを定義します。

「Too Many Tasks」警告

問題：1,000,000 個以上の小さなタスクを作成しています。

解決策：データをより大きな単位に再チャンク化します。dask_array.rechunk() または dask_df.repartition() を使用します。

ベストプラクティス

1. 開発中は常に Dask ダッシュボードを監視して、ボトルネックを特定します。
2. チャンクサイズは慎重に選び、最適なパフォーマンスを得るために、チャンクあたり 100～250MB を目安にします。
3. パフォーマンスとメタデータのサポートを向上させるため、DataFrame には CSV ではなく Parquet 形式を使用します。
4. 再計算を避けるため、複数回再利用する中間結果を永続化します。
5. 計算はできるだけ遅延させ、最終結果が必要な場合にのみ .compute() を呼び出します。
6. Dask DataFrame に対するカスタム pandas 操作には map_partitions を使用します。
7. Dask DataFrame の行を反復処理することは避け、代わりにベクトル化された操作を使用します。
8. 分散クラスターで作業する場合は、共有ストレージ（S3、HDFS、NFS）を使用します。
9. タスクのオーバーヘッドを避けるため、小さなタスクをまとめてバッチ処理します。
10. RAM に収まるデータには Dask を使用しないでください。小規模なデータセットでは pandas／NumPy のほうが高速です。

Dask は、Python を単一スレッドのスクリプト言語から、分散コンピューティングのための世界水準のシステムへと変えます。研究者のノート PC と高性能コンピューティングクラスターをつなぐ架け橋です。