dask

Dask

概要

Dask は、並列計算と分散計算のための Python ライブラリであり、次の 3 つの重要な機能を提供します。

• メモリ容量を超える実行：利用可能な RAM を超えるデータを単一マシンで処理できます
• 並列処理：複数コアにわたる計算速度を向上させます
• 分散計算：複数マシンにわたるテラバイト規模のデータセットに対応します

Dask は、使い慣れた Python API を維持しながら、ノート PC（約 100 GiB の処理）からクラスター（約 100 TiB の処理）までスケールします。

現在のアップストリーム： 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:// paths
uv pip install gcsfs   # gs:// paths

Python 3.10 以降が必要です（3.9 のサポートは 2024.12 に終了）。DataFrame の I/O には PyArrow 16 以降が必要です（dask 2026.1.2 時点）。

このスキルを使用する場面

次の場合にこのスキルを使用してください。

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

主な機能

詳細、コード例、パラメーター表については、references/core-capabilities.md を参照してください。この手順が該当する場合は、こちらをお読みください。

ベストプラクティス

パフォーマンス最適化、メモリ管理戦略、避けるべき一般的な落とし穴についての包括的なガイダンスは、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 (use mode= for overwrite; zarr_array_kwargs for compression)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

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

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. 分散実行にスケールアップして監視する：

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 コンポーネント、操作、パターンに関する詳しい情報が必要な場合は、これらのファイルを読み込んでください。

エージェントの作業手順

1. 環境を確認します。 Python環境とライブラリのバージョン（python -c "import pkg; print(pkg.__version__)"）を確認し、データの形状、型、欠損値を調べます。
2. 入力を特定します。 データまたはユーザーから形式、識別子、パラメーターを確認します。結果に影響する値は推測せず、質問してください。
3. まず小規模なバージョンを実行します。 最初にサンプルまたは単一のフォールドで実行し、実行時間とメモリを確認します。
4. 上記の手順と参考資料を使用して、タスク全体を実行します。
5. 結果を検証します。 保留データ、固定した乱数シード、適切な指標を使用し、データリークがないか確認して、不確実性（CI、シード間の標準偏差）を報告します。
6. 報告します。 実行した内容（バージョン、コマンド、パラメーター）、確認した内容、未確認の事項を明記します。

こうなった場合	対応
メモリ不足、または実行が非常に遅い	サブサンプリングするか、チャンク処理または遅延計算を使用するか、モデルサイズを小さくし、変更内容をユーザーに伝えます。
インストール済みバージョンに、この手順で指定された関数、フラグ、エンドポイントがない	インストール済みバージョンのドキュメント（help()、--help、公式ドキュメント）を確認して適応し、その旨をユーザーに伝えます。APIを決して捏造しないでください。
必須の入力、識別子、またはパラメーターが曖昧	ユーザーに質問するか、実行前に仮定を明示します。

誠実性に関するルール

• 結果、パラメーター、識別子、引用、統計を決して捏造しないでください。実行または検証できない場合は、そのことを明確に伝えます。
• このセッションで計算していない指標を決して報告しないでください。すべての数値について、それを生成したコードの処理経路を示します。
• ここに記載されたバージョン固有の詳細は古くなっている可能性があります。インストール済みバージョンの公式ドキュメントで確認してください。
• 費用が発生する、共有GPUやクラウドのクォータを消費する、個人情報や患者データを扱う、または取り消せない操作を行う前に、必ず確認してください。

関連スキル

• polars：PythonでのETL、分析、pandasからの移行に対応する高性能DataFrameライブラリ。
• ray-data：機械学習ワークロード向けのスケーラブルなデータ処理。
• ray-train：クラスター間の分散トレーニングオーケストレーション。