取得処理が例外なく終わったとき、私たちは成功したと思います。ところが「成功した」が意味しているのは、通信が最後まで通ったことだけです。中身が正しいかどうかは、まだ何も確かめていません。
取得の成功と、データの正しさは別の話です。相手が正常に応答を返していても、その中身が想定と違うことはあります。条件の指定を間違えて別の期間を取っていた、というのは自分側の誤りですが、これも通信は成功します。
実際によくあるのは、件数が普段の半分しかない、特定の日のデータだけ欠けている、同じ識別子が二重に入っている、といった状態です。で時系列として扱うと、こうしたずれは索引の値として静かに残ります。どれも例外を出さずに通過します。そして下流で集計されたあと、数字がおかしいと誰かに言われて初めて気づきます。
だから取得の直後に検算を挟みます。ここで大事なのは、検算のために大掛かりな仕組みを用意しないことです。手間がかかる方法だと、忙しいときに飛ばされます。ファイルに対してそのまま問い合わせを書ければ、数行で済みます。
ファイルに直接 SQL を書く
を使うと、や のファイルに対して直接 を書けます。データベースを立てる必要も、事前に読み込む必要もありません。取得した直後のファイルを指定して、そのまま件数や重複を数えられます。
pandasに読み込んでから調べても同じことはできますが、全部をメモリに載せる必要があります。数百万行のファイルだと、読み込むだけで時間がかかります。検算のためだけに全部読むのは無駄が大きい。
import duckdb
def summarize(path: str) -> None: """ファイルを開かずに、件数と期間と重複を一度に出す。""" con = duckdb.connect() row = con.execute( """ SELECT count(*) AS rows, count(DISTINCT id) AS unique_ids, min(date) AS date_from, max(date) AS date_to, count(*) FILTER (WHERE value IS NULL) AS null_values FROM read_parquet(?) """, [path], ).fetchone()
rows, uniq, d_from, d_to, nulls = row print(f"行数 {rows:,}") print(f"識別子 {uniq:,}（重複 {rows - uniq:,}）") print(f"期間 {d_from} 〜 {d_to}") print(f"値が空 {nulls:,}（{nulls / rows * 100:.1f}%）")何を確かめるか
検算に使う問い合わせは、最初に一度書けば他の取得にも使い回せます。列名を引数にしておけば、対象を変えるだけで同じ検査が動きます。確かめる項目は、データの種類によらずおおむね共通です。前回と比べてどうかを見るのが基本で、絶対値だけを見ても異常だと分かりません。件数1万件が多いのか少ないのかは、前回が2万件だったと分かって初めて判断できます。
| 項目 | 見つかるもの | 判断の基準 |
|---|---|---|
| 行数 | 取りこぼし・重複取得 | 前回比で大きく増減していないか |
| 識別子の重複 | ページングのずれ | 重複がゼロか |
| 期間の端 | 日付のずれ・欠け | 想定した範囲に収まっているか |
| 欠損の割合 | 仕様変更・項目の廃止 | 前回と同水準か |
| 型 | スキーマの変化 | 数値のはずの列が文字列になっていないか |
行数の検査は、上限と下限の両方を見てください。減っていれば取りこぼしを疑いますが、増えすぎている場合も異常です。のずれで同じページを二重に取っていると、行数だけが膨らみます。識別子の重複と合わせて見ると、どちらが起きているか判別できます。
この中でも欠損の割合の変化は見落とされやすく、しかも危険です。ある列が突然すべて空になっていても、行数は変わらないので件数の検査は通ります。配信元がその項目の提供をやめた、というが背景にあることが多い。
前回との比較を残す
前回比を見るには、前回の結果を残しておく必要があります。検算の結果そのものを1行のファイルに追記していくのが単純です。中身のデータではなく、検算の数字だけを残すので、量はほとんど増えません。
import jsonfrom pathlib import Path
LOG = Path("data/checks.jsonl")
def record_and_compare(stats: dict, tolerance: float = 0.3) -> list[str]: """前回の記録と比べて、変化が大きい項目を返す。""" prev = None if LOG.exists(): lines = [ln for ln in LOG.read_text(encoding="utf-8").splitlines() if ln.strip()] if lines: prev = json.loads(lines[-1])
with LOG.open("a", encoding="utf-8") as f: f.write(json.dumps(stats, ensure_ascii=False, default=str) + "\n")
if prev is None: return [] # 初回は比較対象が無い
alerts = [] for key in ("rows", "unique_ids"): before, after = prev.get(key), stats.get(key) if not before or after is None: continue change = abs(after - before) / before if change > tolerance: alerts.append(f"{key}: {before:,} -> {after:,}（{change:.0%} の変化）") return alerts止めれば下流に汚れたデータが流れませんが、復旧するまで全部が止まります。流せば止まりませんが、汚れたデータで集計されます。どちらが良いかはデータの用途で決まります。決めておかないと、事故の最中に判断することになります。
検算は取得の一部だと考える
検算を別の工程にすると、必ず飛ばされます。取得のスクリプトの最後に検算まで含めて、検算まで通って初めて成功という作りにしておくのが確実です。数行で済むので、後から足すより最初から入れておくほうが安上がりです。
検算の結果は、通ったときも記録に残してください。異常が出たときだけ記録する作りだと、平常時の数字が分からなくなります。「今日は1万件だった」を毎日残しておくからこそ、5千件になった日に異常だと判断できます。
データの品質をどう定義して、どこまで自動で守るかについては 前処理の現場 で詳しく扱っています。次回は、応答の形そのものを固定する方法を見ます。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。