AIデータパイプラインの実装：時点整合・データ版固定・学習と推論の特徴量一致を組む手順【2026年版】

AIデータパイプラインは、集計表を作る分析向けのパイプラインと同じ部品で組めますが、壊れ方が違います。学習データに予測時点より後の値が混ざる、学習時と推論時で特徴量の計算がずれる、どのデータで学習したモデルか後から分からなくなる、の3つです。本記事では、DuckDB 2.0のASOF JOINで時点のそろった学習データを作り、学習と推論で同じ変換関数とpanderaの検査を通し、DVCで学習データの版を固定するまでを、コード付きで組み立てます。RAG索引への供給を学習系と分ける理由と、特徴量ストアや外部へ任せる判断基準も示します。版番号はいずれも2026年11月時点のものです。

まとめ：AIデータパイプラインは時点整合と版固定を先に入れ特徴量の一致を測る

最初に入れる仕組みは2つです。特徴量を「使えるようになった時刻」で結合して未来の値を混ぜないことと、学習に使ったデータの版をモデルと同じコミットで固定することです。どちらも後から足すと、過去に作ったモデルの評価をやり直す羽目になります。

学習と推論のずれは、変換関数を1つのモジュールにまとめても、ゼロにはなりません。推論時の特徴量をログに残し、学習側の再計算値と突き合わせる計測を、稼働初日から回してください。モデルが数本でデータが1台に収まるうちはDuckDBとDVCの最小構成で足り、オンライン推論で低遅延の特徴量取得が要る時点で特徴量ストアを検討します。

AIデータパイプラインの出口は学習・推論・RAGの3系統で分析向けと壊れ方が違う

取込・保存・変換・実行制御という層の分け方は、通常のデータパイプラインと変わりません。層ごとの役割と冪等性の原則はデータパイプラインとはで整理しているので、ここではAI向けに出口が増えることで何が変わるかに絞ります。

学習・推論・RAG索引の3つの出口と求められる鮮度・再現性の違いの対応表

AIデータパイプラインの出口は、モデルの学習、推論時の特徴量の受け渡し、RAGの検索索引の3つです。出口ごとに、守るべき性質が異なります。

学習の出口は過去の再現性を、RAG索引は最新への追随を優先します。向きが逆なので、1本のパイプラインに両方を背負わせると、どちらかの要求を落とすことになります。

分析向けパイプラインでは表に出ない未来情報の混入と学習時と推論時のずれ

集計表であれば、昨日の値が今日少し変わっても、利用者は最新の数字を見れば済みます。学習データでは、同じ値の変化が事故につながる原因です。たとえば10月12日時点の解約予測を学習する行に、10月13日に更新された注文件数が結合されると、モデルは本番で手に入らない情報で正解を当てるようになります。検証データでは精度が高いのに、本番で急に外れるというのが、この混入の現れ方です。

GoogleのRules of Machine Learningは、学習時と推論時の性能差をtraining-serving skewと呼び、Rule #31で、学習時と推論時に同じテーブルを結合すると、その間にテーブルの値が変わりうると注意しています。次の章からは、この2つの壊れ方をコードで防ぐ手順に入ります。

検証環境とDuckDBのASOF JOINで未来の値を混ぜない学習データを作る手順

題材は、顧客ごとの解約予測です。正解ラベルの表と特徴量の表を、予測時点より前の値だけで結合します。

duckdb 2.0.0・pandas 3.1系・pandera 0.36系の検証環境

PyPIのduckdbでは、2026年10月21日公開の2.0.0が最新で、Python 3.11以上を要求します。panderaは11月10日公開の0.36.0が最新です。版を固定して入れます。

mkdir -p ai_pipeline/data && cd ai_pipeline
python -m venv .venv
source .venv/bin/activate
pip install "duckdb==2.0.0" "pandas==3.1.1" "pandera==0.36.0" "dvc[s3]"
python -c "import duckdb, pandera; print(duckdb.__version__, pandera.__version__)"

DuckDB 2.0はストレージ形式が1系から変わっています。既存の1系のファイルを読む場合の注意点はDuckDB 2.0とはにまとめました。

特徴量が使えるようになった時刻を持たせるテーブル設計とサンプルデータ

特徴量の表には、集計対象の日付ではなく、その値が使えるようになった時刻を持たせます。日次バッチが翌朝6時に終わるなら、10月14日分の注文件数が使えるのは10月15日6時です。集計日で結合すると、予測時点では存在しなかった値を学習に入れてしまいます。次のbuild_training.pyの前半で、2つの表を作ります。

import duckdb

con = duckdb.connect("data/ai_pipeline.duckdb")

# 正解ラベル：予測した時刻と、その後に解約したかどうか
con.execute("""
create or replace table labels as
select * from (values
    (1, timestamp '2026-10-05 10:00', 1),
    (1, timestamp '2026-10-20 09:00', 0),
    (2, timestamp '2026-10-12 15:00', 1)
) t(customer_id, predict_at, churned)
""")

# 特徴量：available_at は日次バッチが値を書き終えた時刻
con.execute("""
create or replace table features as
select * from (values
    (1, timestamp '2026-10-01 06:00', 3),
    (1, timestamp '2026-10-15 06:00', 7),
    (2, timestamp '2026-10-10 06:00', 2),
    (2, timestamp '2026-10-13 06:00', 9)
) t(customer_id, available_at, orders_30d)
""")

顧客2のorders_30d = 9は、予測時刻の10月12日15時より後に書かれた値です。正しく結合できていれば、学習データには入りません。

ASOF LEFT JOINで予測時点より前の最新値だけを結合するクエリ

DuckDBのFROM句とJOINの公式ドキュメントによると、ASOF JOINは順序を表す列に不等号の条件が1つ必要で、それ以外の条件は等号でなければなりません。右側に該当行が無い左側の行も残したいときはASOF LEFT JOINを使います。build_training.pyの後半は次のとおりです。

training = con.sql("""
select l.customer_id, l.predict_at, l.churned,
       f.available_at, f.orders_30d
from labels l
asof left join features f
  on l.customer_id = f.customer_id
 and l.predict_at >= f.available_at
order by l.customer_id, l.predict_at
""")
training.show()
training.write_parquet("data/train.parquet")

結果は3行です。顧客1の10月5日の行には10月1日の3が、10月20日の行には10月15日の7が付きます。顧客2の行に付くのは10月10日の2で、翌日に書かれた9は入りません。単純に顧客IDで最新の値を結合した場合は、顧客1の10月5日の行にも7が、顧客2には9が付き、どちらも予測時点では知り得なかった値です。ASOF JOINの考え方はDuckDBのAsOf Joinガイドが、株価と保有株数の表を例に結果の行まで示しています。

古すぎる特徴量を欠損扱いにする有効期限の付け方とFeastのTTLとの対応

ASOF JOINは、どれだけ古くても直前の値を結合します。半年前の注文件数を「現在の値」として使いたくない場合は、結合後に経過時間で切る必要があるため、次のコードで条件を指定してください。ASOFの結合条件には等号以外を足せないため、外側で判定します。

select customer_id, predict_at, churned,
       case when predict_at - available_at <= interval 30 day
            then orders_30d end as orders_30d
from read_parquet('data/train.parquet')

この考え方は特徴量ストアにもそのまま対応します。Feastのpoint-in-time joinの解説では、entity dataframeの各行の時刻から、TTLを上限に過去へさかのぼって値を探すと説明しています。TTLの基準はクエリを実行した現在時刻ではなく、各行の時刻です。上のSQLのinterval 30 dayが、FeastのTTLにあたります。

学習と推論で同じ変換関数を呼びpanderaで特徴量の型と範囲を検査する実装

時点をそろえても、学習と推論で特徴量の計算式が違えばモデルは外れます。変換を1か所に集め、出口の手前で検査を挟みます。特徴量の作り方そのものは特徴量エンジニアリングとはを参照してください。

変換処理を1つのPythonモジュールにまとめ学習と推論の両方から読み込む構成

Rules of Machine LearningのRule #32は、学習パイプラインと推論パイプラインでできる限りコードを共有するよう勧め、学習と推論で別の言語を使うとコードの共有が難しくなると述べています。次のfeatures.pyを、学習のバッチと推論のAPIの両方からimportします。

# features.py：学習と推論の両方から呼ぶ唯一の変換処理
import pandas as pd


def build_features(raw: pd.DataFrame) -> pd.DataFrame:
    orders = raw["orders_30d"].fillna(0).astype("int64")
    out = pd.DataFrame(index=raw.index)
    out["orders_30d"] = orders
    # 注文0件のときは平均単価を0とする（0除算を避ける）
    out["avg_amount"] = (raw["amount_30d"] / orders.where(orders > 0)).fillna(0.0)
    out["is_member"] = raw["plan"].eq("member").astype("int8")
    return out

推論側で同じ計算をSQLやJavaScriptで書き直すと、欠損の埋め方や0除算の扱いが少しずつ食い違います。推論APIがPythonで書けない事情があるなら、変換済みの特徴量をバッチで計算して保存し、推論側は読むだけにする構成のほうが、ずれの原因を減らせます。

pandera 0.36系のDataFrameSchemaによる推論前の異常値検査

panderaのDataFrame Schemasの公式ドキュメントは、pandas向けの検査をimport pandera.pandas as paで読み込む書き方で統一しています。Columnは既定で欠損を許さず、lazy=Trueを付けると、最初の違反で止めずにすべての違反をまとめて返します。

# check_features.py
import pandera.pandas as pa

feature_schema = pa.DataFrameSchema({
    "orders_30d": pa.Column("int64", pa.Check.greater_than_or_equal_to(0)),
    "avg_amount": pa.Column(float, pa.Check.greater_than_or_equal_to(0)),
    "is_member": pa.Column("int8", pa.Check.isin([0, 1])),
})


def validated(df):
    # 違反があれば SchemaErrors を送出し、学習にも推論にも渡さない
    return feature_schema.validate(df, lazy=True)

学習ではvalidated(build_features(raw))の結果だけをtrain.parquetに書き、推論でも同じ2つの関数を通してからモデルに渡します。検査の項目が増え、データソースごとにレポートを残したくなったらGreat Expectationsへ移す選択肢があります。外部から集めたデータを学習に入れる場合は、意図的な汚染への備えとしてデータポイズニングの防御策も確認してください。

推論時の特徴量をログに残し学習側の再計算値と突き合わせるずれの計測SQL

Rule #29は、推論時に使った特徴量を記録して学習に回すことが、学習と推論をそろえる最善の方法だとしています。Rule #37で求めているのは、学習と推論の間にあるずれの計測です。推論APIでrequest_idごとに特徴量を縦持ちの表へ書き出し、夜間バッチで同じリクエストの特徴量を再計算して突き合わせます。

select o.feature_name,
       count(*) as n,
       avg(case when abs(o.value - t.value) > 1e-9 then 1 else 0 end) as mismatch_rate
from serving_log o
join recomputed_features t using (request_id, feature_name)
group by o.feature_name
order by mismatch_rate desc

同じ入力から同じ関数で計算しているのにmismatch_rateが0でない特徴量は、計算の実装か参照しているテーブルの時点がずれています。精度が落ちてから原因を探す事態を避ける方法が、日次でこの値を監視の対象に加える運用です。監視の通知の組み方はデータパイプライン監視の実装で解説しています。

DVCで学習データの版を固定しモデルと同じコミットで再現できるようにする手順

半年前に作ったモデルの精度が落ちたとき、同じデータで学習し直せないと、劣化の原因がデータかコードかを切り分けられません。学習用のParquetをGitで扱える形にします。

dvc addとリモート登録で学習用Parquetをバージョン管理するコマンド

DVCのGet Startedによると、dvc addはデータ本体の代わりに小さな.dvcファイルを作り、本体は.gitignoreでGitの管理から外します。本体はリモートのストレージへdvc pushで送ります。

git init && dvc init
git commit -m "DVCを初期化"
dvc remote add -d storage s3://example-ml-bucket/dvcstore
dvc add data/train.parquet
git add data/train.parquet.dvc data/.gitignore .dvc/config
git commit -m "学習データ 2026-11-12版"
git tag train-data-20261112
dvc push

モデルの学習スクリプトとこの.dvcファイルを同じコミットに入れておけば、コミットIDが「どのコードでどのデータを学習したか」の記録になります。バケット名は自社の環境に置き換えてください。

タグとdvc checkoutで過去のモデルを学習したときのデータへ戻す操作

過去の版へ戻すときは、Gitで.dvcファイルを戻してから、dvc checkoutで作業ディレクトリのデータ本体をその版に合わせます。

git checkout train-data-20261112 -- data/train.parquet.dvc
dvc checkout
python train.py

手元のキャッシュに無い版は、先にdvc pullでリモートから取得します。学習のたびにタグを打つ運用なら、モデルの登録簿にタグ名を書くだけで学習データまでたどれる仕組みです。実験管理やモデル登録の道具を含めたMLOpsの全体像はMLOpsツール比較で比べています。

RAG索引向けの取り込みを学習用パイプラインと分けて差分で更新する設計

生成AIの業務利用では、モデルを学習しなくても、社内文書を検索索引へ流す取り込みが必要になります。この取り込みは、学習データとは逆に「常に最新が正しい」出口です。

文書のハッシュで変更分だけを埋め込み直し削除を索引へ伝える管理表

RAG索引は、文書ごとに本文のハッシュ値を管理表に持ち、前回と変わった文書だけをチャンク分割と埋め込みに回します。全件を毎回埋め込み直すと、埋め込みAPIの費用と時間が文書数に比例して増えるためです。元の文書が消えたら、索引からも該当のチャンクを消します。削除の反映漏れは、退職者の資料や廃止された規程を回答に使い続ける形で表に出ます。

学習データのように過去の版を固定する必要はないので、DVCでの版管理はこちらには使いません。取り込み系と推論系の工程の分け方、再索引を冪等に保つ方法はRAGパイプラインとはで、埋め込みモデルの選び方は埋め込みモデルとはで詳しく扱っています。

AIデータパイプラインを自前で組む条件と特徴量ストアや外部へ任せる判断基準

ここまでの構成は、DuckDBのファイルとGitとオブジェクトストレージだけで動きます。どこまでこれで続け、どこから製品や外部の力を借りるかを決めます。

DuckDBとDVCの最小構成で足りる規模と特徴量ストアが要る兆候

次の条件がそろう間は、本記事の最小構成で足ります。

• 推論はバッチで行い、結果を表に書き出して業務システムが読む
• 特徴量を使うモデルが数本で、特徴量の定義を共有する相手がいない
• 学習データが1台のサーバのディスクとメモリに収まる

APIへのリクエストごとに数十ミリ秒で特徴量を返す必要が出たら、特徴量ストアを検討する時期です。Feastの特徴量取得の解説は、学習データの生成にget_historical_features、リアルタイム推論にget_online_featuresを使い分けると説明しています。同じ定義から両方を引けることが、本記事で手作業で守った時点整合と計算の一致を製品側で担保する仕組みです。PyPIのfeastは11月10日公開の0.68.0で、まだ1.0に達していません。0系の間は版の更新で設定の書き方が変わりうる点を、採用時の保守費用に含めてください。

逆に、推論がバッチだけなのに特徴量ストアを入れるのは過剰です。オンラインストアの運用が増えるだけで、ASOF JOINとParquetで済む処理に置き換わるものはありません。

外注するときに社内で決めておく予測時点・正解ラベル・許容するずれ

取込や変換のコード、DVCとストレージの設定、ずれの計測の仕組みは外部に任せられます。社内に残すべきなのは3つの定義です。何時の時点で予測するのか、何をもって正解とするのか（解約なら何日間の未利用で解約とみなすか）、学習と推論のずれをどこまで許すか。いずれも業務の定義で、委託先が決めると、ASOF JOINの結合条件や検査のしきい値が業務の実態とずれたまま固まります。

予測時点と正解ラベルの定義から一緒に整理し、学習データの生成から推論への供給までを組みたい場合は、データ分析基盤構築・MLOps構築支援で相談を受け付けています。

AIデータパイプラインの実装でよくある質問と構成・ツール選びの判断

AIデータパイプラインを設計・実装するときによく出る質問に、構成の判断に関わるものから答えます。

AIデータパイプラインと通常のデータパイプラインは何が違いますか？

部品はほぼ同じで、違いは出口に求められる性質です。分析向けは最新の数字を正しく見せれば足りますが、学習向けは過去の特定時点の状態を再現できること、推論向けは学習時と同じ計算で特徴量を渡すことが求められます。そのため、時点を考慮した結合、学習データの版の固定、学習と推論のずれの計測という、分析向けには無い3つの仕組みが加わります。取込や変換の基本的な組み方はデータパイプライン構築の手順と共通です。

生成AIのAPIを使うだけでもAIデータパイプラインは必要ですか？

モデルを自社で学習しない場合でも、RAGで社内文書を参照させるなら取り込みのパイプラインは必要です。必要なのは本記事のRAG索引の章の部分で、ASOF JOINやDVCによる版固定は要りません。文書の追加・更新・削除を索引へ反映する差分更新と、どの文書がいつ取り込まれたかの記録を優先してください。プロンプトや回答の評価まで含めた運用は、LLMOpsの領域になります。

特徴量ストアのFeastは最初から入れるべきですか？

推論がバッチだけなら、最初から入れる必要はありません。本記事のASOF JOINと共通の変換関数で、時点整合と計算の一致は守れます。リアルタイム推論で低遅延の特徴量取得が必要になったとき、または複数のチームが同じ特徴量を使い回したくなったときが導入を検討する時期です。その時点で特徴量の定義がfeatures.pyのような1か所にまとまっていれば、Feastの特徴量定義へ移す作業も小さく済みます。

学習データの版管理はS3のバージョニングだけで済ませられますか？

オブジェクト単位で過去の版を残すことはできますが、どのモデルがどの版で学習されたかの対応は別に記録する必要があります。DVCを使う利点は、その対応を.dvcファイルとしてGitのコミットに含められる点です。学習コードとデータの版が同じコミットに並ぶので、記録の漏れが起きにくくなります。ファイルが1つで再学習もまれなら、S3のバージョニングと台帳の記録で運用しても構いません。

AIデータパイプラインは最初にどこまで作ればよいですか？

最初の1本は、モデル1つの学習データを時点をそろえて作り、版を固定するところまでで十分です。推論がバッチなら、同じfeatures.pyと検査を通して推論用の表を作る処理を足します。ずれの計測は、推論のログが溜まり始める稼働初日から入れてください。実行の定期化や上流の更新を合図に動かす仕組みは、データパイプライン自動化の実装の手順で後から足せます。

関連記事

• データパイプラインとは？種類・構成要素と冪等性から考える設計判断を解説【2026年版】：AI向けに限らない設計の原則と構成要素
• RAGパイプラインとは？取り込み系と推論系に分ける工程設計と再索引の実装判断：RAG索引への供給を工程単位で設計する方法
• データパイプライン構築の手順：dltで取込・dbtで変換・Parquetで配信する最小構成とコストの増え方【2026年版】：学習データの手前にある取込と変換の組み方
• データパイプライン監視の実装：鮮度・件数・スキーマ変化をdbtとAirflowで検知する手順【2026年版】：学習と推論のずれを監視に載せるときの通知設計
• MLOpsツール比較｜実験管理・パイプライン・監視の主要スタックと選定基準：データの版管理と組み合わせる実験管理・モデル登録の道具