過去の区間を再実行する — 二重に数えないように

目標

区間を受け取って過去をもう一度回す実行ツールrunner.pyを作ります。定期実行と同じコードでパーティション単位の置換を行い、区間の予約で定期実行との衝突を防ぎ、すでに出力した区間は封印して訂正として残し、累計集計がバックフィルのために2倍になることを自分で再現してから、直します。

なぜ重要なのか

バックフィルは、過去の区間をもう一度回すことですが、難しいのは計算ではなく調整です。バックフィルが回っている間も、定期実行は回ります。2つが同じ日付に同時に書き込むと、どちらが勝ったのか誰にもわかりません。同じコースの冪等のラボは、同じ行を2回入れても結果が同じになるようにすることでした。ここは、行ではなく区間です。パーティションの表が冪等でも、累計の表が冪等でなければ、バックフィル1回で数字が膨らみます。足す累計は、バックフィルに出会うと必ず間違え、導出する累計は、何回回しても同じ値を出します。そして、戻ってはいけない場所があります。すでに送ったレポートとすでに出たアラートは、データではなく出来事です。その区間は封印し、新しい値が出たら、上書きする代わりに訂正として別に記録して初めて、「そのとき私たちが何を送ったか」と「今何が正しいか」を、両方言えます。採点ツールは、提出された文言を信じません。一時ディレクトリに、採点ツールが作ったパーティションを用意して、環境変数でそのストアを指し、自分で作った実行ツールを実際に実行して、sqliteの表を直接開いて突き合わせます。日付と金額は、実行ごとに変わります。

ステップ

/root/backfill/gen_events.pyを作成して実行し、/root/backfill/eventsの下に、dt=2026-02-01.jsonlからdt=2026-02-14.jsonlまで、14日分のパーティションを作ってください。
/root/backfill/runner.pyにinitとrun <시작일> <끝일> <주인>(プレースホルダーは、開始日、終了日、オーナーです)を作り、区間のパーティションを数えてdailyに入れさせてください。
runがパーティションを丸ごと置換するように直し、丸ごと回した答えと、1日ずつ回した答えと、再実行の結果を、/root/backfill/split.jsonに書いてください。
rollup-add <시작일> <끝일>(プレースホルダーは、開始日と終了日です)とrollupを追加し、バックフィルのあとに2つの方式がどれだけ開くかを、/root/backfill/double.jsonに書いてください。
claim <시작일> <끝일> <주인>とrelease <시작일> <끝일> <주인>(プレースホルダーは、開始日、終了日、オーナーです)を追加して、区間を予約させてください。
runが、他者が押さえているパーティションをスキップして、skippedとして報告するようにしてください。スキップしたものがあれば、終了コードは5です。
seal <시작일> <끝일>とamend <날짜> <사유>(プレースホルダーは、開始日、終了日、日付、理由です)を追加してください。封印されたパーティションはrunが触らず、遅れて届いた伝票は、correctionsに訂正として残ります。
/root/backfill/backfill_report.jsonと/root/backfill/backfill_report.mdで、1枚を残してください。

参考

実行の契約: python3 /root/backfill/runner.py <명령> ...(プレースホルダーはコマンドです)。答えは、JSONの1つの塊として標準出力に出します。終了コードは、0が成功、2が知らないコマンドか引数の数が違う、3が状態ストアがない、4が予約が拒否された、5がスキップした区間がある、6が封印されていない区間にamendを呼んだ、です。
状態ストアのパスは環境変数BACKFILL_DBで、元のパーティションのフォルダはBACKFILL_EVENTSで、変えられる必要があります。デフォルト値は、それぞれ/root/backfill/state.dbと/root/backfill/eventsです。採点ツールが、この2つの変数で自分のストアを指します。
パーティションファイルの名前はdt=YYYY-MM-DD.jsonlで、行の形は{"order_id": 문자열, "dt": 날짜, "amount": 정수, "status": "paid" 또는 "refund"}です(プレースホルダーは、文字列、日付、整数です。statusはpaidとrefundのどちらかを意味し、その間の韓国語の語は「または」という意味です)。返金の行のamountは負の数です。1つのパーティションのordersは行数、amountはamountの合計です。
表は5つです。daily(dt, orders, amount, owner)・rollup(metric, value)・claims(dt, owner)・seals(dt)・corrections(dt, orders_before, amount_before, orders_after, amount_after, reason)。dailyには主キーをかけません。置換を自分で実装してみることが、このラボの要点です。
runの応答: {"owner": 문자열, "done": [날짜...], "skipped": [[날짜, 사유]...], "orders": 정수, "amount": 정수}(プレースホルダーは、文字列、日付、理由、整数です)。ordersとamountは、doneに入ったパーティションだけを足した値です。理由は、claimed_by:<주인>(プレースホルダーはオーナーです)・sealed・no_dataの3つです。元のファイルがない日付(no_data)は、終了コードを5にしません。
claimの応答: {"owner": 문자열, "claimed": [날짜...], "denied": [[날짜, 주인]...]}(プレースホルダーは、文字列、日付、オーナーです)。自分がすでに押さえた日付は、claimedにそのまま入ります。releaseの応答: {"owner": 문자열, "released": [날짜...]}(プレースホルダーは、文字列、日付です)で、自分の予約だけを解除します。
sealの応答: {"sealed": [날짜...]}(プレースホルダーは日付です)。amendの応答: {"dt": 날짜, "orders_before": 정수, "amount_before": 정수, "orders_after": 정수, "amount_after": 정수, "reason": 문자열}(プレースホルダーは、日付、整数、文字列です)。amendは、dailyを直さず、correctionsに1行を加えます。
rollup-addの応答とrollupの応答は{"mode": "add" 또는 "derive", "order_total": 정수, "amount_total": 정수}です(プレースホルダーは、整数です。modeはaddとderiveのどちらかを意味し、その間の韓国語の語は「または」という意味です)。rollup-addは、区間の合計を既存の値に足し、rollupは、daily全体から計算し直して上書きします。
性能で判定しません。バックフィルがどれだけかかったかは測らず、何をどの順序で触ったかだけを残してください。
公式ドキュメント: python sqlite3・SQLite Transaction・SQLite UPSERT・Airflow Dag Runs
よくある間違い: バックフィル用のコードを別に作る、区間全体を1つのトランザクションに入れる、他者の区間を黙って上書きする、累計を足す方式のままにする、すでに出力した区間を上書きして、そのとき何を送ったかを失う。
状態ストアを目で見るには、sqlite3 /root/backfill/state.db 'SELECT * FROM daily ORDER BY dt'を使ってください。

14日分のパーティションを作る

/root/backfill/gen_events.pyを作成して実行し、/root/backfill/eventsの下に、dt=2026-02-01.jsonlからdt=2026-02-14.jsonlまで作ってください。1日が1ファイルで、返金の行のamountは負の数です。

日付はdatetime.date.fromisoformatとtimedeltaで増やせばよいです。1日ごとに行数を変えておくと、あとでパーティションごとの合計を区別できます。返金の行をいくつか混ぜて、金額が単純な累計にならないようにしてください。

区間を受け取って回す実行ツールを作る

/root/backfill/runner.pyにinitとrun <시작일> <끝일> <주인>(プレースホルダーは、開始日、終了日、オーナーです)を作ってください。runは、区間のパーティションごとに行数と金額を数えてdailyに入れ、元のファイルがない日付は、skippedにno_dataとして残します。

定期実行は、区間が1日の場合にすぎません。そのため、関数は1つだけ作ってください。表は、initでCREATE TABLE IF NOT EXISTSで作り、dailyには主キーをかけません。no_dataは、終了コードを5にしません。

パーティション単位で置換する

runが1つのパーティションをもう一度回すときに、そのパーティションの古い結果を消してから入れるように直してください。そのあと、丸ごと回した答えと、1日ずつ回した答えと、再実行の結果を、/root/backfill/split.jsonにwhole・by_day・rerun_changedとして書き、/root/backfill/state.dbのdailyも、日付ごとに1行になるように、埋め直してください。

置換とは、その日付の行を消して入れ直すことです。丸ごと回した答えと、1日ずつ回した答えを比べるには、ストアを2回新しく作る必要がありますが、BACKFILL_DB環境変数で一時ストアを指せば、本番のストアに触れずに比較できます。

累計集計が2倍になることを再現する

rollup-add <시작일> <끝일>(プレースホルダーは、開始日と終了日です)とrollupを追加してください。そのあと、一時ストアで全区間を回して累計に足し、一部の区間をバックフィルしてから、その区間をもう一度足して、導出した値とどれだけ開くかを、/root/backfill/double.jsonにbackfill_range・true_total・add_after_backfill・gapとして書いてください。

rollup-addは、区間の合計を既存の値に足し、rollupは、daily全体を数え直して上書きします。開いた幅は、ちょうどバックフィル区間の金額の合計と同じである必要があります。その区間が、累計に2回入ったからです。

区間を先に予約する

claim <시작일> <끝일> <주인>とrelease <시작일> <끝일> <주인>(プレースホルダーは、開始日、終了日、オーナーです)を追加してください。他者が押さえている日付は、deniedにオーナーと一緒に入れて、終了コードは4です。releaseは、自分の予約だけを解除します。

claims表の主キーを日付にしておけば、1つの日付を2人のオーナーが押さえることはできません。自分がすでに押さえた日付をもう一度押さえることは、拒否ではなく成功にしてください。リトライが失敗に見えれば、誰もリトライしません。

他者の区間はスキップして報告する

runが、他者が押さえているパーティションに触れず、skippedにclaimed_by:<주인>(プレースホルダーはオーナーです)として残すようにしてください。スキップしたものがあれば、終了コードは5です。自分が押さえたパーティションは、そのまま回します。

待たず、上書きもしないことが要点です。スキップして伝えれば、呼び出した側が、もう一度呼ぶか、人を呼ぶかを決められます。予約の一覧は、区間を回る前に1回だけ読めばよいです。

すでに出力した区間は訂正として残す

seal <시작일> <끝일>とamend <날짜> <사유>(プレースホルダーは、開始日、終了日、日付、理由です)を追加してください。封印されたパーティションは、runがsealedとしてスキップします。そのあと、2026-02-02のパーティションに、遅れて届いた伝票を1行足して、その日付を封印したうえで、amendで訂正を残してください。訂正のbeforeとafterは、異なる必要があります。

amendは、dailyを直しません。元データから数え直した値と、dailyに残っている値を、並べてcorrectionsに書くだけです。封印されていない日付にamendを呼んだら、終了コード6で拒否してください。そのような日付は、そのままrunすればよいからです。

再実行の1回分を1枚で報告する

/root/backfill/backfill_report.jsonに、partitions・orders・amount・sealed・corrections・rollup・double_gapを書き、/root/backfill/backfill_report.mdに、## 무엇을 다시 돌렸나、## 정기 실행과 어떻게 부딪혔나、## 누적 집계는 왜 두 배가 되나、## 되돌리면 안 되는 자리の4つの節で書いてください(韓国語の見出しで、順に「何をもう一度回したか」「定期実行とどうぶつかったか」「累計集計はなぜ2倍になるのか」「戻してはいけない場所」を意味します)。

partitions・orders・amountはdaily表から、sealedはsealsから、correctionsは行数で、rollupは導出した累計から読みます。double_gapは、ステップ4で測った値をそのまま使います。レポートには、開いた金額を数字で書いてください。