処理履歴を参照し、新規・更新ファイルだけを取り込むPython ETL

Python / pandas / SQLite / SQL / pytest

課題

外部システムから定期的に配置されるCSVを毎回すべて取り込むと、処理済みファイルを繰り返し読み込み、無駄な処理が増え、処理済みの範囲も追いにくくなります。既存ETLが持つ対象判定の業務ロジックをPythonとSQLへ移すことをテーマにしました。

元データ

架空の商品コードと金額を持つAAA.csv、BBB.csv、CCC.csvを用意しました。SQLiteには前回成功時のファイル更新時刻と処理時刻・行数を保存します。実際の企業名、テーブル名、実データ、機密情報は使用していません。

Pythonでどう処理するか

ファイルのメタデータを探索し、SQLiteの履歴と比較します。AAAは同じ更新日時なのでSKIP、BBBはinputの方が新しいためPROCESS、CCCは履歴がないためPROCESS。探索・履歴SQL・分類・対象一覧・CSV取り込みを分離しました。日時比較はナノ秒単位の整数、表示はtimezone-awareなUTC日時を使用し、SQLの値はパラメータ化しています。

出力物

理由付きの処理対象一覧と、SQLiteへの取り込み結果を出力します。更新されたファイルの既存行は置き換え、成功時だけ履歴を更新します。データと履歴を同じトランザクションで確定し、失敗時は両方を戻します。再実行では成功済みファイルがスキップされます。

発注者にとってのメリット

CSVの内容を読み込む対象を新規・更新ファイルへ絞り、繰り返し処理を減らします。判定理由と成功履歴から処理状況を確認できます。既存ETLのルールを読み解き、Python / SQLへ責務を分けて移行できることを示しています。削減時間は未計測です。

サンプルの範囲と実務経験

Pentaho ETLからPython ETLへの置き換え業務で扱った処理パターンを、機密情報や実データを含まない形で再構成しました。元記事のコードをそのまま転載せず、自動テストと失敗時の履歴管理を追加しています。単一フォルダ・単一ワーカーが前提です。更新日時を維持した内容変更、並列実行、大規模CSVの分割読み込み、業務キーでの更新・削除は本番仕様として別途確認が必要です。