alterlab-dask

Dask

概要

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

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

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

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

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

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

該当しない場合

シナリオ	代わりに使用
データが RAM に収まり、単一マシン上で DataFrame を最大限高速に処理したい場合	alterlab-polars
クラスターを使わず、単一マシンで 10 億行規模のデータをメモリ外で探索し、ビッグデータをプロットする場合	alterlab-vaex
データの計算ではなく、チャンク化・圧縮された Zarr ストアとコーデックを設計する場合	alterlab-zarr

バージョンに関する注意

2025.1 以降、クエリプランニングを行う DataFrame（旧 dask-expr）が唯一の Dask DataFrame 実装となり、dask に同梱されています。dask.config.set({"dataframe.query-planning": False}) は利用できなくなりました。テキスト列は string dtype（Dask では PyArrow をバックエンドとし、pandas 3 では str）として読み込まれます。また、集約処理でこれらの列が暗黙にスキップされることはありません。非数値列がある場合、ddf.mean() は TypeError を、ddf.groupby(...).mean() は NotImplementedError を発生させるため、列を選択するか numeric_only=True を渡してください。

主な機能

Dask には、用途に応じて使い分けられる 5 つの主要コンポーネントがあります。

1. DataFrame — 並列 pandas 処理

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

使用する場面：

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

リファレンスドキュメント：Dask DataFrame の詳しい使い方については、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(numeric_only=True).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
• 配列やデータフレームの形式に適合しない Python オブジェクトの処理
• メモリ効率の良いストリーミング処理が必要な場合

リファレンスドキュメント：Dask Bags の包括的なガイダンスについては、以下を含む references/bags.md を参照してください。

• テキストファイルと JSON ファイルの読み込み
• 関数型の操作（map、filter、fold、groupby）
• データフレームへの変換
• 一般的なパターン（ログ分析、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()

重要なポイント：

• 最初のデータクレンジングに使用し、その後データフレーム／配列に変換します
• パフォーマンスを向上させるには、groupby の代わりに foldby を使用します
• 操作はストリーミング方式で、メモリ効率に優れています
• 複雑な操作には、構造化形式（データフレーム）に変換します

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

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

使用する場面：

• 動的に変化するワークフローの構築
• タスクの即時実行が必要な場合（遅延実行ではない）
• 計算が実行時の条件に依存する場合
• カスタムの並列アルゴリズムの実装
• ステートフルな計算が必要な場合

リファレンスドキュメント：Dask Futures の包括的なガイダンスについては、以下を含む references/futures.md を参照してください。

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

簡単な例：

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. Schedulers - 実行バックエンド

目的：Dask タスクの実行方法と実行場所（スレッド、プロセス、分散）を制御します。

スケジューラーの選び方：

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

リファレンスドキュメント：Dask Schedulers の包括的なガイダンスについては、以下を含む 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(numeric_only=True).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. Named aggregation keeps flat string column names;
# a dict-of-lists .agg() creates MultiIndex columns that Parquet rejects.
summary = ddf.groupby('category').agg(
    amount_sum=('amount', 'sum'),
    amount_mean=('amount', '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(numeric_only=True).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 コンポーネント、操作、パターンについて詳細な情報が必要な場合は、これらのファイルを読み込んでください。