「あれ、どこでデータ消えたんだ...？」を防ぐdagster asset check【競馬AI開発#62】

この【競馬AI開発】シリーズでは、競馬予想AIを作ることを通して、機械学習・データサイエンスが学べるコンテンツをお届けしています。今回のテーマは、dagster の asset checks を使い、「データを見張る仕組み」を加えることです。

‍■今回やること
‍前回までで、dagster の「materialize」ボタンを押せば競馬AIに必要なデータが組み上がり、保存される仕組みが整いました。処理が成功すればデータは手に入ります。

ただ、成功したからといって、そのデータが「正しい」とは限りません。列の値がおかしくても、レースが欠けていても、データ処理自体は何事もなく完了します。
その場合、ずっと後になって「なぜかデータの行数が減っている」「なぜかモデルの精度が悪化している」という形で問題に気づくことになりますが、全ての処理は一見"正しく完了している"ように見えるため、どの部分におかしなデータが潜んでいるか分からず、骨の折れる調査をする羽目になるわけです。

今回導入する asset checks は、この問題を解決します。アセットを materialize するたびに検査が自動で走り、アセットごとの合否と理由が dagster の UI に並びます。‍

壊れたデータには blocking をかけられます。検査に引っかかったアセットの下流は実行されないので、問題のあるデータが黙って下流に流れることがなくなります。‍このような「データを見張る仕組み」を追加するのが、今回の目標です。

‍■競馬AIについて
‍netkeiba.comから収集したデータを機械学習モデルに学習させることで期待リターンを計算し、リターンが最大化するように自動投票を行う競馬予想AIを開発しています。

すでにシステムとしては一通り完成しており、現在はこの完成したシステムをベースに、特徴量の追加やモデルの改善など、さらなる精度向上に取り組んでいる段階です。

この【競馬AI開発】シリーズでは、こうしたシステムをゼロから構築する過程を、コード付きで詳しく解説しています。ソースコードは、下に進んでダウンロード・解凍すれば使うことができるので、ぜひ実際にコードを動かしてみてください。

1. asset checksで「データの正しさ」を見張る

たとえば、あるテーブルに本来1000レースぶんのデータが入るはずなのに、800レースしか入っていなかったとします。行数が少なくても、列の型や値の範囲が正しくても、処理自体はエラーなく終わります。

asset checks は、この「気づけない問題」をアセット単位で検査する仕組みです。アセットの materialize が完了するたびに、そのアセットに紐づいた検査が自動で実行されます。検査の結果（合格か、不合格か、なぜ不合格なのか）は dagster の UI にアセットごとに表示されるので、データを作った直後に「想定通りのデータが出来ているか」を確認できます。

dagster の公式ドキュメントでは、asset checks について次のように説明されています。

では、実際にどう書くのか。まずは簡単なデモを見てみましょう。

‍■asset checkの基本形
‍次のコードは、小さな DataFrame を返すアセットと、そのデータを検査するアセットチェックの最小構成です。

import dagster as dg
import pandas as pd


@dg.asset
def sales_summary() -> pd.DataFrame:
    """日別の売上合計。"""
    return pd.DataFrame({
        "date": ["2025-01-01", "2025-01-02", "2025-01-03"],
        "amount": [12000, 8500, 15000],
    })


@dg.asset_check(asset=sales_summary, name="no_negative_amount", blocking=True)
def check_no_negative(sales_summary: pd.DataFrame) -> dg.AssetCheckResult:
    """売上金額が負でないことを検査する。"""
    negatives = sales_summary[sales_summary["amount"] < 0]
    return dg.AssetCheckResult(
        passed=negatives.empty,
        severity=dg.AssetCheckSeverity.ERROR,
        metadata={
            "total_rows": len(sales_summary),
            "negative_rows": len(negatives),
        },
    )


defs = dg.Definitions(
    assets=[sales_summary],
    asset_checks=[check_no_negative],
)

‍@dg.asset_check デコレータの引数を見ていきます。

asset=sales_summary で、このチェックがどのアセットを検査するかを指定しています。これにより、dagster は「このチェックは sales_summary に紐づくものだ」と認識し、sales_summary を materialize するたびにこのチェックが自動で走ります。

name="no_negative_amount"‍ はチェックの名前です。ひとつのアセットに複数の検査を紐づけることができるので、名前で区別します。

blocking=True は「このチェックが失敗したら、下流のアセットを止める」という指定です。これについては後ほど詳しく説明します。

関数の中身はシンプルです。引数でアセットの DataFrame を受け取り、検査ロジックを書いて、AssetCheckResult を返します。passed に合否の真偽値を渡し、severity でチェックが失敗したときの深刻度を指定し、metadata に検査の詳細情報を付けます。metadata に入れた値は UI から確認でき、何が起きたのかを把握するのに役立ちます。

このコードで dagster を起動して sales_summary を materialize すると、アセットの生成に続いてチェックが自動で走り、UI にその結果が表示されます。

‍■わざと壊して失敗させる
‍では、データをわざと壊してみましょう。amount に ‍-500 を含むデータを返すように書き換えます。

@dg.asset
def sales_summary() -> pd.DataFrame:
    return pd.DataFrame({
        "date": ["2025-01-01", "2025-01-02", "2025-01-03"],
        "amount": [12000, -500, 15000],  # 2日目がマイナス
    })

この状態で materialize すると、チェックは不合格になります。severity が ERROR なので UI にはエラーが表示されます。そして blocking=True を指定してあるので、もし sales_summary の下流にアセットがあれば、そのアセットの materialize は実行されません。壊れたデータがそのまま次の処理に渡ることがないのです。

‍■severity ― 止めるべき壊れ方と、警告に留める壊れ方
‍AssetCheckResult の severity には ERROR と WARN の二段階があります。

ERROR は「確実に壊れている。下流に流してはいけない」ときに使います。WARN は「気になるが、致命的ではない」ときに使います。

この二つは blocking と組み合わさることで、パイプラインの制御に関わってきます。

blocking=True  + severity=ERROR  → 下流を止める
blocking=True  + severity=WARN   → 下流は止めない（警告だけ記録）
blocking=False + severity=ERROR  → 下流は止めない（結果はUIに表示）
blocking=False + severity=WARN   → 下流は止めない（結果はUIに表示）

blocking=True と severity=ERROR の両方がそろったときだけ、下流がブロックされます。‍この設計のおかげで、「止めるほどではないが気づきたい検査」と「止めなければ危険な検査」を同じ仕組みの中で使い分けられます。

ここからは、この仕組みを競馬AIのパイプラインに実際に組み込んでいきます。