dask は何をしますか？
概要
Daskは、並列・分散コンピューティングのためのPythonライブラリであり、以下の3つの重要な機能を提供します。
- メモリ超過の実行: 利用可能なRAMを超えるデータを単一マシンで処理
- 並列処理: 複数のコアを活用した計算速度の向上
- 分散計算: 複数マシンにわたるテラバイト規模のデータセットをサポート
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以上が必要です (3.9のサポートは2024.12で終了)。DataFrame I/OにはPyArrow 16以上が必要です (dask 2026.1.2以降)。
このスキルを使用すべき場合
このスキルは以下の場合に使用します:
- 利用可能なRAMを超えるデータセットを処理する
- pandasやNumPyの操作をより大きなデータセットにスケールさせる
- パフォーマンス向上のために計算を並列化する
- 複数のファイル (CSV、Parquet、JSON、テキストログ) を効率的に処理する
- タスクの依存関係を持つカスタム並列ワークフローを構築する
- ワークロードを複数のコアまたはマシンに分散する
コア機能
Daskは5つの主要なコンポーネントを提供し、それぞれ異なるユースケースに対応します:
1. DataFrame - 並列pandas操作
目的: 並列処理を通じてpandas操作をより大きなデータセットにスケールします。
使用すべき場合:
- 表形式データが利用可能なRAMを超えている
- 複数のCSV/Parquetファイルを一緒に処理する必要がある
- pandas操作が遅く、並列化が必要
- pandasプロトタイプから本番環境へのスケールアップ
リファレンスドキュメント: Dask DataFrameに関する包括的なガイダンスについては、references/dataframes.mdを参照してください。これには以下が含まれます:
- データの読み込み (単一ファイル、複数ファイル、グロブパターン)
- 一般的な操作 (フィルタリング、groupby、結合、集計)
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を使用 - 他のソースから構造化データを扱う場合は、早期にpandas DataFrameに変換
2. Array - 並列NumPy操作
目的: ブロック化アルゴリズムを使用して、NumPyの機能をメモリを超えるデータセットに拡張します。
使用すべき場合:
- 配列が利用可能なRAMを超えている
- NumPy操作の並列化が必要
- 科学データセット (HDF5、Zarr、NetCDF) を扱う
- 並列線形代数や配列操作が必要
リファレンスドキュメント: Dask Arrayに関する包括的なガイダンスについては、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()
重要なポイント:
- チャンクサイズは重要 (1チャンクあたり約100 MBを目安に)
- 操作はチャンク単位で並列に実行
- 効率的な操作のために必要に応じてデータを再チャンク化
- Daskにない操作には
map_blocksを使用
3. Bag - 非構造化データの並列処理
目的: 関数型操作を使用して、非構造化データや半構造化データ (テキスト、JSON、ログ) を処理します。
使用すべき場合:
- テキストファイル、ログ、JSONレコードの処理
- 構造化分析の前のデータクリーニングとETL
- 配列やデータフレーム形式に適合しないPythonオブジェクトを扱う
- メモリ効率の良いストリーミング処理が必要
リファレンスドキュメント: Dask Bagに関する包括的なガイダンスについては、references/bags.mdを参照してください。これには以下が含まれます:
- テキストファイルとJSONファイルの読み込み
- 関数型操作 (map、filter、fold、groupby)
- DataFrameへの変換
- 一般的なパターン (ログ分析、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に変換
- パフォーマンス向上のために
groupbyの代わりにfoldbyを使用 - 操作はストリーミング処理され、メモリ効率が良い
- 複雑な操作には構造化形式 (DataFrame) に変換
4. Future - タスクベースの並列化
目的: タスクの実行と依存関係を細かく制御しながら、カスタムの並列ワークフローを構築します。
使用すべき場合:
- 動的で進化するワークフローを構築する
- 即時のタスク実行が必要 (遅延ではない)
- 計算が実行時の条件に依存する
- カスタム並列アルゴリズムを実装する
- 状態を持つ計算が必要
リファレンスドキュメント: Dask Futureに関する包括的なガイダンスについては、references/futures.mdを参照してください。これには以下が含まれます:
- 分散クライアントのセットアップ
- タスクの送信とFutureの操作
- タスクの依存関係とデータ移動
- 高度な調整 (キュー、ロック、イベント、アクター)
- 一般的なパターン (パラメータスイープ、動的タスク、反復アルゴリズム)
クイック例:
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を使用する前に、以下を検討してください:
- より良いアルゴリズム
- 効率的なファイル形式 (CSVの代わりにParquet)
- コンパイルコード (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()
# 正しい: すべての計算を一度に
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
# Extract: データを読み込み
ddf = dd.read_csv('raw_data/*.csv')
# Transform: クリーニングと処理
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])
# Load: 集計して保存
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コンポーネントを選択してください:
データタイプ:
- 表形式データ → DataFrame
- 数値配列 → Array
- テキスト/JSON/ログ → Bag (その後DataFrameに変換)
- カスタムPythonオブジェクト → Bag または Future
操作タイプ:
- 標準的なpandas操作 → DataFrame
- 標準的なNumPy操作 → Array
- カスタム並列タスク → Future
- テキスト処理/ETL → Bag
制御レベル:
- 高レベル、自動 → DataFrame/Array
- 低レベル、手動 → Future
ワークフロータイプ:
- 静的計算グラフ → DataFrame/Array/Bag
- 動的、進化する → Future
統合に関する考慮事項
ファイル形式
- 効率的: Parquet、HDF5、Zarr (カラム型、圧縮、並列処理に適切)
- 互換性ありだが遅い: CSV (最初の取り込みのみに使用)
- Array向け: 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コードでスレッドを使用している (プロセスに切り替え)
- データ依存関係が並列化を妨げている