dask

Dask

概要

Dask は、並列処理と分散コンピューティングのための Python ライブラリで、次の 3 つの重要な機能を提供します。

• メモリに収まらないデータの処理: 利用可能な RAM を超えるデータを単一マシン上で実行可能
• 並列処理: 複数のコアにまたがって計算を行い、計算速度を向上
• 分散計算: 複数のマシンにまたがってテラバイト規模のデータセットを処理

Dask は、ノートPC（約 100 GiB のデータを処理）からクラスター（約 100 TiB のデータを処理）まで規模を拡張でき、使い慣れた Python API をそのまま利用できます。

このスキルを使用するタイミング

このスキルは次の場合に使用します。

• 利用可能な RAM を超えるデータセットを処理するとき
• pandas や NumPy の処理をより大きなデータセットにスケールさせるとき
• 計算を並列化してパフォーマンスを向上させるとき
• 複数のファイル（CSV、Parquet、JSON、テキストログ）を効率的に処理するとき
• タスクの依存関係を持つカスタムの並列ワークフローを構築するとき
• 複数のコアやマシンにワークロードを分散させるとき

主要な機能

Dask には 5 つの主要コンポーネントがあり、それぞれ異なるユースケースに適しています。

1. DataFrames - 並列 Pandas 処理

目的: 並列処理によって pandas の処理をより大きなデータセットにスケールさせます。

使用するタイミング:

• 表形式データが利用可能な RAM を超えるとき
• 複数の CSV/Parquet ファイルをまとめて処理する必要があるとき
• pandas の処理が遅く、並列化が必要なとき
• pandas のプロトタイプから本番環境へ拡張するとき

リファレンスドキュメント: Dask DataFrames の詳細なガイドについては、references/dataframes.md を参照してください。この文書には次の内容が含まれます。

• データの読み込み（単一ファイル、複数ファイル、glob パターン）
• 一般的な操作（フィルタリング、groupby、結合、集計）
• map_partitions を使ったカスタム操作
• パフォーマンス最適化のヒント
• 一般的なパターン（ETL、時系列、複数ファイル処理）

簡単な例:

import dask.dataframe as dd

# Read multiple files as single DataFrame
ddf = dd.read_csv('data/2024-*.csv')

# Operations are lazy until 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

# Create large array with chunks
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Operations are lazy
y = x + 100
z = y.mean(axis=0)

# Compute result
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）
• DataFrame への変換
• 一般的なパターン（ログ分析、JSON 処理、テキスト処理）
• パフォーマンスに関する考慮事項

簡単な例:

import dask.bag as db
import json

# Read and parse JSON files
bag = db.read_text('logs/*.json').map(json.loads)

# Filter and transform
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# Convert to DataFrame for analysis
ddf = processed.to_dataframe()

重要なポイント:

• 最初のデータクレンジングに使用し、その後 DataFrame または Array に変換します。
• パフォーマンスを高めるには groupby の代わりに foldby を使用します。
• 操作はストリーミング方式で、メモリ効率が良いです。
• 複雑な操作には、構造化フォーマット（DataFrame）に変換して処理します。

4. Futures - タスクベースの並列化

目的: タスクの実行と依存関係をきめ細かく制御しながら、カスタムの並列ワークフローを構築します。

使用するタイミング:

• 動的に変化するワークフローを構築する場合
• タスクを遅延させず、すぐに実行する必要がある場合
• 計算が実行時の条件に依存する場合
• カスタムの並列アルゴリズムを実装する場合
• 状態を保持する計算が必要な場合

リファレンスドキュメント: Dask Futures の包括的なガイドについては、references/futures.md を参照してください。内容は次のとおりです:

• 分散クライアントのセットアップ
• タスクの送信と future の操作
• タスクの依存関係とデータ移動
• 高度な協調機能（キュー、ロック、イベント、アクター）
• 一般的なパターン（パラメータスイープ、動的タスク、反復アルゴリズム）

簡単な例:

from dask.distributed import Client

client = Client()  # Create local cluster

# Submit tasks (executes immediately)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# Gather results
results = client.gather(futures)

client.close()

重要なポイント:

• 分散クライアントが必要です（単一マシンの場合でも同様）
• タスクは送信されるとすぐに実行されます
• 繰り返しの転送を避けるため、大きなデータは事前に scatter してください
• タスクごとに約 1 ms のオーバーヘッドがかかるため、数百万の小さなタスクには適していません
• ステートフルなワークフローにはアクターを使用してください

5. スケジューラー - 実行バックエンド

目的: Dask のタスクをどのように、どこで実行するかを制御します（スレッド、プロセス、分散）。

スケジューラーの選択基準:

• スレッド（デフォルト）: NumPy/Pandas の操作、GIL を解放するライブラリ、共有メモリの利点がある場合
• プロセス: 純粋な Python コード、テキスト処理、GIL に制約される操作
• 同期: pdb によるデバッグ、プロファイリング、エラーの把握
• 分散: ダッシュボード、複数マシンのクラスター、高度な機能が必要な場合

リファレンスドキュメント: Dask スケジューラーの包括的なガイドについては、references/schedulers.md を参照してください。内容は次のとおりです:

• 各スケジューラーの詳細な説明と特性
• 設定方法（グローバル、コンテキストマネージャー、計算ごと）
• パフォーマンスの考慮事項とオーバーヘッド
• 一般的なパターンとトラブルシューティング
• 最適なパフォーマンスのためのスレッド設定

簡単な例:

import dask
import dask.dataframe as dd

# Use threads for DataFrame (default, good for numeric)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # Uses threads

# Use processes for Python-heavy work
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Use synchronous for debugging
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Can use pdb

# Use distributed for monitoring and scaling
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Uses distributed with dashboard

重要なポイント:

• スレッド: オーバーヘッドが最も小さく（約 10 µs/タスク）、数値計算に最適です
• プロセス: GIL を回避でき（約 10 ms/タスク）、Python の処理に最適です
• 分散: 監視ダッシュボードを備え（約 1 ms/タスク）、クラスターへスケールできます
• 計算ごと、またはグローバルにスケジューラーを切り替えられます

ベストプラクティス

包括的なパフォーマンス最適化のガイド、メモリ管理の戦略、避けるべき一般的な落とし穴については、references/best-practices.md を参照してください。主要な原則は次のとおりです:

まずはよりシンプルな解決策を検討する

Dask を使う前に、次の点を検討してください:

• より良いアルゴリズム
• 効率的なファイル形式（CSV の代わりに Parquet）
• コンパイル済みコード（Numba、Cython）
• データのサンプリング

重要なパフォーマンスのルール

1. データをローカルで読み込んでから Dask に渡さない

# Wrong: Loads all data in memory first
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correct: Let Dask handle loading
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. compute() の繰り返し呼び出しを避ける

# Wrong: Each compute is separate
for item in items:
    result = dask_computation(item).compute()

# Correct: Single compute for all
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)  # Monitor performance, identify bottlenecks

一般的なワークフローのパターン

ETL パイプライン

import dask.dataframe as dd

# Extract: Read data
ddf = dd.read_csv('raw_data/*.csv')

# Transform: Clean and process
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load: Aggregate and save
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

非構造化データから構造化データへのパイプライン

import dask.bag as db
import json

# Start with Bag for unstructured data
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Convert to DataFrame for structured analysis
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

大規模な配列計算

import dask.array as da

# Load or create large array
x = da.from_zarr('large_dataset.zarr')

# Process in chunks
normalized = (x - x.mean()) / x.std()

# Save result
da.to_zarr(normalized, 'normalized.zarr')

カスタムの並列ワークフロー

from dask.distributed import Client

client = Client()

# Scatter large dataset once
data = client.scatter(large_dataset)

# Process in parallel with dependencies
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Gather results
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 (for numeric data)
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: 高度なクラスター管理と監視

デバッグと開発

反復的な開発ワークフロー

1. 同期スケジューラーで小さなデータをテストする:

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Can use pdb, easy debugging

2. サンプルデータでスレッドを使って検証する:

sample = ddf.head(1000)  # Small sample
# Test logic, then scale to full dataset

3. 監視のために distributed でスケールさせる:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitor performance
result = computation.compute()

よくある問題

メモリエラー:

• チャンクサイズを小さくする
• persist() を戦略的に使用し、完了したら削除する
• カスタム関数でメモリリークがないか確認する

起動が遅い:

• タスクグラフが大きすぎる（チャンクサイズを大きくする）
• map_partitions や map_blocks を使ってタスク数を減らす

並列化の効率が悪い:

• チャンクが大きすぎる（パーティション数を増やす）
• Python コードでスレッドを使用している（プロセスに切り替える）
• データの依存関係により並列実行が妨げられている

リファレンスファイル

すべてのリファレンスドキュメントは、詳細な情報が必要に応じて読み込めます。

• references/dataframes.md - Dask DataFrame の完全ガイド
• references/arrays.md - Dask Array の完全ガイド
• references/bags.md - Dask Bag の完全ガイド
• references/futures.md - Dask Futures と分散コンピューティングの完全ガイド
• references/schedulers.md - スケジューラーの選択と設定の完全ガイド
• references/best-practices.md - 包括的なパフォーマンス最適化とトラブルシューティング

ここで示した簡単なガイドを超えて、特定の Dask コンポーネント、操作、パターンについて詳しい情報が必要な場合に、これらのファイルを読み込んでください。