ml-data-pipeline-architecture

ML データパイプラインのアーキテクチャ

Polars、Arrow、ClickHouse を使用した効率的な ML データパイプラインのパターンです。

ADR: 2026-01-22-polars-preference-hook（効率性に関する設定フレームワーク）

注: PreToolUse フックによって Polars の使用が優先されます。Pandas を使用するには、ファイルの先頭に # polars-exception: <reason> を追加します。

自己進化型スキル: このスキルは使用を通じて改善されます。指示に誤りがある場合、パラメーターが変化した場合、または回避策が必要になった場合は、先延ばしにせず、すぐにこのファイルを修正してください。実際に再現可能な問題がある場合にのみ更新してください。

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

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

• データパイプラインで Polars と Pandas のどちらを使うか判断するとき
• ゼロコピーの Arrow パターンを使ってメモリ使用量を最適化するとき
• ClickHouse から PyTorch DataLoader にデータを読み込むとき
• 大規模データセット向けに遅延評価を実装するとき
• 既存の Pandas コードを Polars に移行するとき

---

1. 判断ツリー：Polars と Pandas

Dataset size?
├─ < 1M rows → Pandas OK (simpler API, richer ecosystem)
├─ 1M-10M rows → Consider Polars (2-5x faster, less memory)
└─ > 10M rows → Use Polars (required for memory efficiency)

Operations?
├─ Simple transforms → Either works
├─ Group-by aggregations → Polars 5-10x faster
├─ Complex joins → Polars with lazy evaluation
└─ Streaming/chunked → Polars scan_* functions

Integration?
├─ scikit-learn heavy → Pandas (better interop)
├─ PyTorch/custom → Polars + Arrow (zero-copy to tensor)
└─ ClickHouse source → Arrow stream → Polars (optimal)

---

2. ゼロコピーのパイプラインアーキテクチャ

Pandas の問題点

# BAD: 3 memory copies
df = pd.read_sql(query, conn)     # Copy 1: DB → pandas
X = df[features].values           # Copy 2: pandas → numpy
tensor = torch.from_numpy(X)      # Copy 3: numpy → tensor
# Peak memory: 3x data size

Arrow による解決策

# GOOD: 0-1 memory copies
import clickhouse_connect
import polars as pl
import torch

client = clickhouse_connect.get_client(...)
arrow_table = client.query_arrow("SELECT * FROM bars")  # Arrow in DB memory
df = pl.from_arrow(arrow_table)                          # Zero-copy view
X = df.select(features).to_numpy()                       # Single allocation
tensor = torch.from_numpy(X)                             # View (no copy)
# Peak memory: 1.2x data size

---

3. ClickHouse 統合パターン

パターン A：Arrow ストリーム（推奨）

def query_arrow(client, query: str) -> pl.DataFrame:
    """ClickHouse → Arrow → Polars (zero-copy chain)."""
    arrow_table = client.query_arrow(f"{query} FORMAT ArrowStream")
    return pl.from_arrow(arrow_table)

# Usage
df = query_arrow(client, "SELECT * FROM bars WHERE ts >= '2024-01-01'")

パターン B：Polars ネイティブ（よりシンプル）

# Polars has native ClickHouse support (see pola.rs for version requirements)
df = pl.read_database_uri(
    query="SELECT * FROM bars",
    uri="clickhouse://user:pass@host/db"
)

パターン C：Parquet エクスポート（バッチジョブ）

# For reproducible batch processing
client.query("SELECT * FROM bars INTO OUTFILE 'data.parquet' FORMAT Parquet")
df = pl.scan_parquet("data.parquet")  # Lazy, memory-mapped

---

4. PyTorch DataLoader との統合

最小限の変更パターン

from torch.utils.data import TensorDataset, DataLoader

# Accept both pandas and polars
def prepare_data(df) -> tuple[torch.Tensor, torch.Tensor]:
    if isinstance(df, pd.DataFrame):
        df = pl.from_pandas(df)

    X = df.select(features).to_numpy()
    y = df.select(target).to_numpy()

    return (
        torch.from_numpy(X).float(),
        torch.from_numpy(y).float()
    )

X, y = prepare_data(df)
dataset = TensorDataset(X, y)
loader = DataLoader(dataset, batch_size=32, pin_memory=True)

カスタム PolarsDataset（大規模データ）

class PolarsDataset(torch.utils.data.Dataset):
    """Memory-efficient dataset from Polars DataFrame."""

    def __init__(self, df: pl.DataFrame, features: list[str], target: str):
        self.arrow = df.to_arrow()  # Arrow backing for zero-copy slicing
        self.features = features
        self.target = target

    def __len__(self) -> int:
        return self.arrow.num_rows

    def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]:
        row = self.arrow.slice(idx, 1)
        x = torch.tensor([row[f][0].as_py() for f in self.features], dtype=torch.float32)
        y = torch.tensor(row[self.target][0].as_py(), dtype=torch.float32)
        return x, y

---

5. 遅延評価のパターン

パイプラインの合成

# Define transformations lazily (no computation yet)
pipeline = (
    pl.scan_parquet("raw_data.parquet")
    .filter(pl.col("timestamp") >= start_date)
    .with_columns([
        (pl.col("close").pct_change()).alias("returns"),
        (pl.col("volume").log()).alias("log_volume"),
    ])
    .select(features + [target])
)

# Execute only when needed
train_df = pipeline.filter(pl.col("timestamp") < split_date).collect()
test_df = pipeline.filter(pl.col("timestamp") >= split_date).collect()

大容量ファイルのストリーミング

# Process file in chunks (never loads full file)
def process_large_file(path: str, chunk_size: int = 100_000):
    reader = pl.scan_parquet(path)

    for batch in reader.iter_batches(n_rows=chunk_size):
        # Process each chunk
        features = compute_features(batch)
        yield features.to_numpy()

---

6. スキーマの検証

設定に Pydantic を使用

from pydantic import BaseModel, field_validator

class FeatureConfig(BaseModel):
    features: list[str]
    target: str
    seq_len: int = 15

    @field_validator("features")
    @classmethod
    def validate_features(cls, v):
        required = {"returns_vs", "momentum_z", "atr_pct"}
        missing = required - set(v)
        if missing:
            raise ValueError(f"Missing required features: {missing}")
        return v

DataFrame のスキーマ検証

def validate_schema(df: pl.DataFrame, required: list[str], stage: str) -> None:
    """Fail-fast schema validation."""
    missing = [c for c in required if c not in df.columns]
    if missing:
        raise ValueError(
            f"[{stage}] Missing columns: {missing}\n"
            f"Available: {sorted(df.columns)}"
        )

---

7. パフォーマンスのベンチマーク

操作	Pandas	Polars	高速化
CSV の読み込み（1GB）	45s	4s	11x
行のフィルタリング	2.1s	0.4s	5x
グループ化と集計	3.8s	0.3s	13x
ソート	5.2s	0.4s	13x
メモリ使用量のピーク	10GB	2.5GB	4x

ベンチマーク：5,000 万行、20 列、MacBook M2

---

8. 移行チェックリスト

フェーズ 1：Arrow サポートの追加

• 依存関係に polars = "<version>" を追加します（PyPIを参照）
• データクライアントに query_arrow() を実装します
• メモリプロファイラーでゼロコピーを検証します

フェーズ 2：エントリーポイントで Polars を使用

• トレーナーのエントリーポイントに pl.from_pandas() ラッパーを追加します
• prepare_sequences() が両方の型を受け入れるように更新します
• 変換後にスキーマ検証を追加します

フェーズ 3：遅延評価への完全移行

• ファイル読み込みを pl.scan_* に変換します
• 変換を遅延評価で合成します
• .to_numpy() の直前にのみ .collect() を呼び出します

---

9. 避けるべきアンチパターン

禁止：不必要に API を混在させる

# BAD: Convert back and forth
df_polars = pl.from_pandas(df_pandas)
df_pandas_again = df_polars.to_pandas()  # Why?

禁止：早すぎる段階で collect する

# BAD: Defeats lazy evaluation
df = pl.scan_parquet("data.parquet").collect()  # Full load
filtered = df.filter(...)  # After the fact

# GOOD: Filter before collect
df = pl.scan_parquet("data.parquet").filter(...).collect()

禁止：メモリ不足を無視する

# BAD: Loads entire file
df = pl.read_parquet("huge_file.parquet")

# GOOD: Stream in chunks
for batch in pl.scan_parquet("huge_file.parquet").iter_batches():
    process(batch)

---

参考資料

• Polars ユーザーガイド
• Polars 移行ガイド
• Apache Arrow Python
• ClickHouse Python クライアント
• PyTorch データ読み込み
• Polars Preference Hook ADR

---

トラブルシューティング

問題	原因	解決策
読み込み中にメモリ使用量が急増する	早すぎるタイミングでデータを収集している	遅延評価を使い、必要な場合にのみ collect() を呼び出します
Arrow への変換に失敗する	サポートされていないデータ型	object 列を確認し、ネイティブ型に変換します
ClickHouse の接続エラー	ポートまたは認証情報が間違っている	host:8123（HTTP）または host:9000（ネイティブ）を確認します
ゼロコピーが機能しない	中間で pandas に変換している	to_pandas() の呼び出しを削除し、Arrow/Polars のままにします
Polars フックがコードをブロックする	例外を指定せずに pandas を使用している	ファイル先頭に # polars-exception: reason コメントを追加します
group-by 操作が遅い	大規模データセットに pandas を使用している	5～10 倍の高速化のために Polars に移行します
スキーマの検証に失敗する	列名の大文字と小文字が区別される	ソースの列名と完全に一致しているか確認します
PyTorch DataLoader でメモリ不足が発生する	データセット全体をメモリに読み込んでいる	遅延アクセスのために Arrow をバックエンドに使用する PolarsDataset を使います
Parquet のスキャン性能が低い	述語プッシュダウンを使用していない	遅延評価のために collect() の前にフィルターを追加します
テンソルの型が一致しない	Float64 と Float32 が一致していない	numpy の前に .cast(pl.Float32) で明示的にキャストします

実行後の振り返り

このスキルの完了後、タスクを終了する前に振り返ります。

0. 現在地を確認します。 — 編集する前に、この SKILL.md の正式なパスを見つけます。
1. 何が失敗しましたか？ — 原因となった指示を修正します。
2. 予想以上にうまくいったことは何ですか？ — 推奨される実践として取り入れます。
3. 何が実態とずれていましたか？ — 現状に合わなくなったスクリプト、参照先、依存関係を修正します。
4. 記録します。 — きっかけ、修正、根拠を記載した進化ログのエントリーを残します。

先送りしてはいけません。次回の呼び出しでは、ここに残したものが引き継がれます。