ml-data-pipeline

ML データパイプライン

概要

データの取り込み、検証、前処理、特徴量エンジニアリング、データセットのバージョン管理、特徴量ストア、バッチおよびストリーミングのパイプライン、データ品質の監視には、このスキルを使用してください。ML では、データパイプラインの正確性はモデルの高度さよりも重要なことがよくあります。パイプラインは、リークを防いだ学習データと、一貫性のあるサービング特徴量を生成する必要があります。

データパイプラインの不変条件

• 生データは不変であるか、スナップショットで参照できるようにします。
• スキーマ、統計情報、品質要件は、学習とサービングの前に検証します。
• 変換にはバージョンを付け、再現可能にします。
• オーバーサンプリング、ターゲットエンコーディング、特徴量選択、正規化など、リークの原因になりやすい処理の前にデータを分割します。
• 時間依存の特徴量には、予測時点で利用可能な情報のみを使用します。
• オフラインの学習特徴量とオンラインのサービング特徴量を一致させます。
• 機密データは最小限に抑え、アクセスを制御し、暗号化し、監査します。

取り込みとストレージ

データの形状とアクセスパターンに基づいてストレージを選択します。表形式のバッチ ML では、Parquet/Arrow を使用したオブジェクトストレージが有力なデフォルトです。Delta Lake、Apache Iceberg、または Hudi を使うと、ACID テーブル、スキーマ進化、タイムトラベルが可能になります。管理された SQL 特徴量にはデータウェアハウス、埋め込み検索にはベクトルストア、オンラインの行動データにはストリーミングログを使用します。生データ、クリーンデータ、特徴量、モデル用データのレイヤーは個別に保存します。ADF が Azure ML コードアセットのバージョンを渡すために使用する Azure Storage ポインターブロブには、ml-azureml-adf-automation を読み込みます。

大規模なデータセットでは、列指向形式、時間または上位ドメインによるパーティション分割、圧縮、述語プッシュダウン、マニフェストファイルを優先してください。小さなファイルが大量にできないようにし、必要に応じてコンパクションします。すべての学習実行にデータセットのスナップショット識別子を記録します。

Feast 特徴量ストアの設計

Feast は、オンラインとオフラインの特徴量の一貫性を維持するモジュール型の特徴量ストアです。

1. 特徴量ストアの設定（feature_store.yaml）

project: fraud_detection
registry: data/registry.db
provider: local
online_store:
  type: redis
  connection_string: "localhost:6379"
offline_store:
  type: file

2. 特徴量の定義（features.py）

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, ValueType
from feast.types import Float32, Int64

user = Entity(name="user_id", value_type=ValueType.INT64, join_keys=["user_id"])

user_transactions_source = FileSource(
    path="data/user_transactions.parquet",
    event_timestamp_column="timestamp",
    created_timestamp_column="created_timestamp",
)

user_transactions_fv = FeatureView(
    name="user_transactions_feature_view",
    entities=[user],
    ttl=timedelta(days=90),
    schema=[
        Field(name="transaction_count_30d", dtype=Int64),
        Field(name="total_amount_30d", dtype=Float32),
    ],
    online=True,
    source=user_transactions_source,
)

処理エンジン

エンジン	最適な用途
pandas	小〜中規模のインメモリ探索とシンプルなパイプライン
Polars	高速なローカル／遅延列指向処理、pandas では扱いきれない規模のワークロード
Spark	大規模な分散 ETL、レイクハウスのワークフロー、SQL と MLlib の統合
Dask	Python ネイティブの分散配列／データフレームとカスタムワークロード
Ray Data	Ray Train/Tune/Serve と統合された ML 中心の分散前処理
Beam/Flink/Spark Streaming	ストリーミングまたはバッチ／ストリーム統合データフロー
Airflow/Prefect/Dagster	オーケストレーション、スケジューリング、再試行、リネージ、依存関係の管理

高パフォーマンスの Polars 遅延集約パイプライン

Polars の遅延評価では、述語プッシュダウンと射影プッシュダウンを使用して実行計画を最適化します。

import polars as pl

def compute_rolling_user_features(transactions_path: str):
    lazy_df = (
        pl.scan_parquet(transactions_path)
        # Cast timestamp for windowing
        .with_columns(pl.col("timestamp").str.strptime(pl.Datetime))
        .sort("timestamp")
        # Define rolling calculation window
        .group_by_dynamic(
            index_column="timestamp",
            every="1d",
            period="30d",
            group_by="user_id"
        )
        .agg([
            pl.col("amount").count().alias("transaction_count_30d"),
            pl.col("amount").sum().alias("total_amount_30d"),
            pl.col("amount").mean().alias("avg_amount_30d")
        ])
        .filter(pl.col("user_id").is_not_null())
    )
    # Collect executes the query optimization and loads results into memory
    return lazy_df.collect()

Spark Dataframe の最適化レシピ

データの偏りや過剰なシャッフルのオーバーヘッドなど、分散学習でよく起きるボトルネックを防ぎます。

1. 偏りのある結合を防ぐソルティング

from pyspark.sql import functions as F

# Adding a salt column to distribute skewed key values evenly across partitions
skewed_df = skewed_df.with_columns(
    (F.rand() * 10).cast("int").alias("salt")
)
lookup_df = lookup_df.with_columns(
    F.explode(F.array([F.lit(i) for i in range(10)])).alias("salt")
)

# Join on key AND salt to distribute the workload
joined_df = skewed_df.join(lookup_df, ["join_key", "salt"], "inner").drop("salt")

2. ルックアップテーブルのブロードキャスト結合

from pyspark.sql.functions import broadcast

# Explicitly broadcast small dimension dataframe to executors to avoid shuffling large fact table
optimized_joined = large_fact_df.join(broadcast(small_lookup_df), "entity_id", "inner")

データ検証

取り込み時、特徴量生成時、学習時、サービング時に検証します。スキーマ、型、範囲、null、ユニーク性、重複、カテゴリ値の範囲、カーディナリティ、ラベル分布、タイムスタンプの単調性、参照整合性、テキスト／画像／音声の妥当性、埋め込みのノルムを確認します。ツールには Great Expectations、TensorFlow Data Validation、Deequ、pandera、dbt tests、カスタムアサーションがあります。

品質チェックでは、契約違反があれば即座に失敗させ、調査が必要な分布の変化については警告します。検証レポートは学習アーティファクトとともに保存します。本番環境では、生の特徴量と変換後のモデル入力の両方を監視します。

特徴量エンジニアリング

特徴量エンジニアリングは、予測時点に合わせる必要があります。時系列データでは、正しいカットオフ、遅延、到着の遅いデータへの対処を考慮して、ローリングウィンドウを計算します。ターゲットエンコーディングでは、交差検証の各フォールド内でエンコーダーを学習し、平滑化を使用します。カテゴリ特徴量では、カーディナリティとモデルの種類に応じて、ネイティブのカテゴリサポート、ワンホットエンコーディング、ハッシング、埋め込み、ターゲットエンコーディングから選択します。テキストでは、トークナイザーと語彙をバージョン管理します。画像や音声では、前処理パラメーターとデータ拡張を保存します。

リークを防ぐには、「モデルが予測を行う時点で、この特徴量を利用できますか？」と確認します。利用できない場合は、その特徴量を除外するか、ターゲットと予測時点を再設計します。

DVC（Data Version Control）の取り込みワークフロー

DVC は、軽量なメタデータファイルを通じて大規模データセットを Git コミットに紐付け、リポジトリの肥大化を防ぎます。

1. ストレージを初期化して追加する

dvc init
dvc remote add -d myremote s3://my-dvc-bucket/raw-data

2. 新しいデータセットのバージョンを追跡する

# Add dataset to DVC tracking (creates data.parquet.dvc)
dvc add data/raw_transactions.parquet

# Commit DVC metadata file to Git
git add data/raw_transactions.parquet.dvc data/.gitignore
git commit -m "Track transactions dataset v1.0.0 via DVC"

# Push raw binaries to remote cloud storage
dvc push

3. 別のワーカーで追跡対象のバージョンを取得する

git pull
dvc pull

ストリーミングとオンラインパイプライン

ストリーミング ML パイプラインでは、イベント時刻の処理、ウォーターマーク、重複排除、順序付けの戦略、遅れて到着したデータの扱い、exactly-once または at-least-once のセマンティクス、再実行可能性が必要です。オンラインの特徴量更新と、学習ラベルの生成は分離します。特徴量ロジックを変更した場合に備え、永続ログからバックフィルまたは再実行できる経路を確保します。

セキュリティとプライバシー

機密フィールドを最小限に抑え、必要に応じて識別子をトークン化またはハッシュ化し、必要な場合に限って結合可能な状態を維持します。データレイヤーごとにアクセス制御を適用します。実験トラッカーやモデル成果物に生の PII をエクスポートしないようにします。データの保持ポリシーと削除ポリシーを文書化します。

よくあるパイプラインの障害

• 前処理の実装が別々であることによる、学習時とサービング時のずれ。
• ローリング特徴量、集計値、ターゲットエンコーダーにおける未来情報のリーク。
• 文字列、null、カテゴリが誤って変換される、気付かれにくいスキーマのドリフト。
• 学習データと検証データにおけるエンティティの重複。
• 分割前のオーバーサンプリングや正規化。
• バージョン管理されていないデータセットによる、再現不可能な学習。
• 学習用にバックフィルできないストリーミング特徴量。

参考資料

• Great Expectations のドキュメント：https://docs.greatexpectations.io/
• TensorFlow Data Validation のドキュメント：https://www.tensorflow.org/tfx/data_validation
• Feast のドキュメント：https://docs.feast.dev/
• DVC のドキュメント：https://dvc.org/doc
• Apache Spark MLlib と SQL のドキュメント：https://spark.apache.org/docs/latest/
• Polars のドキュメント：https://docs.pola.rs/