ETL テスト：行の来歴・除外・変換後の値を確認する

執筆：Ai Flowi · 公開日 2026-10-11

ETL テストでは、抽出した記録が合意した取り決めに従って変換され、読み込み後の出力または明示的な除外記録に漏れなく含まれることを確認します。「ETL テストとは何か」を実務的に理解するには、行の総数だけでなく、元の識別子、変換後の値、却下理由を比較します。出力の行数が正しくても、1件を重複させて別の1件を失っている可能性があります。

ETLの用語解説はパイプラインの各段階を説明します。このガイドは、6件の記録を照合する演習と、提供されたファイルを確認するPython標準ライブラリのチェックを示します。データウェアハウスを設定したり、テスト済みのビルダーパイプラインを確立したりはしません。

1. 変換の取り決めを明記する

この架空のバッチでは、来歴に一意のsource_idを使い、受け入れた業務記録に一意のitem_keyを使います。規則では項目キーの先頭のゼロを保持し、前後の空白を除き、十進数字だけから成る数量を受け付けます。確認用の表現では数量を非負整数にします。空のキーと数字以外を含む数量は、具体的な理由を付けて除外します。

この規則は意図的に限定されています。符号付きの値、小数の数量、桁区切りには対応しません。それらが自分のデータで有効なら、別の取り決めを定義してください。データクリーニングのリファレンスは、提案した修正が正当かどうかという別の問いを扱います。

2. 元データ・受け入れデータ・除外データのCSVをコピーする

以下のすべての行は、学習用に作成した架空のテストデータです。元の記録は提供された入力です。受け入れ記録と除外記録は、作成した期待値であり、観測したワークフローの出力ではありません。CSVはテキストを保存するため、以下のチェックは数量の正規形のテキストを確認してから整数に変換します。

元データのCSV

source_id,item_key,quantity_raw
S01,0011,2
S02,0012,0
S03, 0013 ,007
S04,0014,two
S05,,3
S06,0016,4

受け入れデータのCSV

source_id,item_key,quantity
S01,0011,2
S02,0012,0
S03,0013,7
S06,0016,4

除外データのCSV

source_id,reason
S04,quantity_not_digits
S05,missing_item_key

S02は、この取り決めでゼロが有効であることを示します。S03は、独立した2つの変換を示します。キーの前後の空白は消えますが、先頭のゼロは残ります。一方、数量のテキスト007は整数の正規形7になります。S04は値を受け入れ出力に含められなくても、来歴を保持します。

3. 行単位のアサーション例を実行する

3つのブロックをsource.csv、accepted.csv、exclusions.csvとして保存し、以下のコードを同じ作業ディレクトリのetl-check.pyに保存します。Pythonの通常モードで実行してください。最適化を有効にするとアサーションが無効になります。ファイルは小さく、表示したとおりのヘッダーを持つ管理されたテストデータです。このコードは提供したスナップショットを確認するもので、サービスを呼び出したり、出力先に読み込んだりしません。

import csv
import re
from collections import Counter
from copy import deepcopy

def read(name):
    with open(name, encoding="utf-8", newline="") as stream:
        return list(csv.DictReader(stream))

def verify(source, accepted, excluded):
    ids = [row["source_id"] for row in source]
    assert all(ids) and len(ids) == len(set(ids)), "source identity"
    actual = [row["source_id"] for row in accepted + excluded]
    assert Counter(actual) == Counter(ids), "lineage mismatch"
    keys = [row["item_key"] for row in accepted]
    assert len(keys) == len(set(keys)), "duplicate item key"
    expected = {}
    rejected = {}
    for row in source:
        sid = row["source_id"]
        key = row["item_key"].strip()
        raw = row["quantity_raw"]
        if not key:
            rejected[sid] = "missing_item_key"
        elif not re.fullmatch(r"[0-9]+", raw):
            rejected[sid] = "quantity_not_digits"
        else:
            expected[sid] = (key, int(raw))
    assert {row["source_id"] for row in accepted} == set(expected), "disposition"
    for row in accepted:
        raw = row["quantity"]
        assert re.fullmatch(r"0|[1-9][0-9]*", raw), "quantity format"
        quantity = int(raw)
        assert type(quantity) is int, "quantity type"
        assert (row["item_key"], quantity) == expected[row["source_id"]], "value mismatch"
    assert {row["source_id"]: row["reason"] for row in excluded} == rejected, "exclusion reason"

if __name__ == "__main__":
    source = read("source.csv")
    accepted = read("accepted.csv")
    excluded = read("exclusions.csv")
    verify(source, accepted, excluded)
    missing = accepted[:-1]
    duplicate = deepcopy(accepted)
    duplicate[1] = deepcopy(duplicate[0])
    assert len(duplicate) + len(excluded) == len(source)
    wrong_value = deepcopy(accepted)
    wrong_value[2]["quantity"] = "8"
    duplicate_key = deepcopy(accepted)
    duplicate_key[1]["item_key"] = duplicate_key[0]["item_key"]
    negatives = [
        (missing, "lineage mismatch"),
        (duplicate, "lineage mismatch"),
        (wrong_value, "value mismatch"),
        (duplicate_key, "duplicate item key"),
    ]
    for bad, reason in negatives:
        try:
            verify(source, bad, excluded)
        except AssertionError as error:
            assert str(error) == reason
        else:
            raise AssertionError("negative fixture accepted")
    wrong_reasons = deepcopy(excluded)
    wrong_reasons[0]["reason"], wrong_reasons[1]["reason"] = wrong_reasons[1]["reason"], wrong_reasons[0]["reason"]
    try:
        verify(source, accepted, wrong_reasons)
    except AssertionError as error:
        assert str(error) == "exclusion reason"
    else:
        raise AssertionError("wrong exclusion reasons accepted")
    print("Synthetic fixture accepted; five negative fixtures rejected")

行の重複ケースでは、S02を別のS01で置き換えます。受け入れ4行と除外2行の合計は依然として6行なので、件数だけのテストは通ります。Counterの比較は出現回数を保持し、重複したS01と欠けたS02を検出して拒否します。集合の比較だけでは「それぞれ正確に1回」という要件を表せません。

4. 除外と、出力先の証拠をレビューする

判定とともに、バッチ参照、元データのスナップショット、規則の改訂版、出力先のスナップショットを記録します。このテストデータでは、受け入れ4行と除外2行で、元の6件の識別子をすべて説明できます。ただし、この照合は1つの境界にすぎません。読み込みを完了と呼ぶ前に、実際の出力先を受け入れファイルと比較してください。

候補ファイルにS06があっても、出力先にS06がなければ未解決のままにします。フィルターや不完全なアクセスによって出力先のエクスポートで記録が隠れる場合は、欠けた識別子を解釈する前に、その対象範囲の問題を解決します。読み戻しの失敗は、先行する書き込みが拒否された証拠にはなりません。

元データの証拠の保持と判断の改訂については、ETLの却下と修正の規則に従ってください。

この演習で、元のIDを変えずにS04とS05の理由を入れ替えます。両方の行が引き続き説明されるため来歴チェックは通りますが、除外理由の比較は失敗する必要があります。S04の数量は数字以外を含み、S05には項目キーがないからです。また、1行が両方の規則に違反する場合の優先順位も定義します。このチェックはキーの欠損を先に報告します。代わりに数量の失敗を報告する期待値なら、別の取り決めをテストしていることになります。記録を分割・集約する場合は、複数の出力を1つの元の行に押し込めず、この1対1の識別子の取り決めを、元データと出力の明示的な対応付けに置き換えてください。

5. ETLテストで最初に失敗した比較をどう特定しますか？

調査中は、失敗したファイルを変更せずに保持します。生成側の規則を直すか、認可された元データの修正を取得し、別の改訂版を生成して再び比較します。期待する受け入れCSVを欠陥のある結果に合わせて編集すると、これをテストにしている独立した期待値を失います。

業務規則が変わり、小数の数量を許容するなら、新しい規則の改訂版を記録し、数量のアサーションを意図的に置き換えます。旧来の数字だけの取り決めによる結果は、以前のテストの説明として引き続き有効です。適格性の規則が異なることを明記せずに、その除外件数を改訂後のバッチと合算してはいけません。

6. AI Flowi Workflowで同じ取り決めを確認する

テーブル確認の状態：文書化済み。Load CSV、Data → DataFrame、Table Operationsはカタログに記載されていますが、この照合についてAi Flowiが本番検証したものではありません。この確認で使うインターフェースの境界には、読み込み、変換、Table Operationsの取り決めを参照してください。

設定例であり、記録されたテストではありません

1. 利用を許可されたサンプルのエクスポートと、Workflowエディターへのアクセスを使います。この確認にモデルや外部出力先のアカウントは不要です。プロバイダーAPIを追加する場合は、自分のアカウントやキーと、別途プロバイダーへの支払いが必要です。
2. Load CSVのcsv_stringに元データのブロックを設定し、代わりとなる他の入力は未設定にします。解析した識別子と、元の数量の文字列を確認します。
3. 確認したオブジェクトをData → DataFrameのdata_listに渡し、出力のdataframeのTableを確認してから、Table Operationsのdfに接続します。
4. 並べ替えて確認する場合は、operationをSort、column_nameをsource_id、ascendingをtrueに設定します。Tableであるoutputを確認してください。並べ替えは確認の補助であり、Pythonの変換や除外チェックを実装するものではありません。

想定する出力は、確認可能なテーブルと、外部での照合の判定です。カタログは、この読み込みコンポーネントの型変換規則や、組み合わせた設定の実行成功を証明しません。抽出や読み込みに別のエンドポイントが関わる場合は、APIの取り決めのチェックを使ってください。この小さなチェックは、ウェアハウスの自動検証、同時読み込み、増分の照合を必要とする大規模データセットには適しません。

関連情報：リバースETL とは：業務でのデータ活用と書き込みの境界。