日々のデータ同期で「全部取り直すと時間がかかる」「前回どこまで処理したか分からなくなる」「失敗時に二重登録や欠落が起きる」といった悩みはよくあります。本記事では、CSV／外部API／データベースを実務で少ないコストで同期するための差分同期（インクリメンタルETL）の手順を、Pythonの最小限パターンで示します。エンジニアでなくても追えるように、関数・辞書・ファイル入出力・例外処理の基本を復習しながら進めます。
1) 本記事で解く問題と適用範囲
対象は、小〜中規模の業務データ（数万行程度まで）で、毎回全件を取り直すコストが重い場合。想定する同期の条件を表にまとめます。
| 項目 | 想定値／方針 |
|---|---|
| ソース | CSV / Googleスプレッドシート / 外部API |
| ターゲット | CSV / SQLite / REST API（アップサート対応） |
| 頻度 | バッチ（毎時〜毎日） |
| 整合性要件 | 最終的整合性を許容。重要なトランザクションは分離して扱う |
2) 差分検出の基本パターン
差分検出でよく使う方法を比較します。
| 方法 | 長所 | 短所 | 適用例 |
|---|---|---|---|
| タイムスタンプ | 実装が簡単（updated_at） | 時計ずれ／タイムゾーン問題に注意 | DBのupdated_at、APIのmodified |
| ハッシュ（行単位） | 内容の変更を確実に検出 | 全行計算が必要、コストあり | CSVの内容比較 |
| シーケンスID | 増分取得が容易 | 欠番や並行挿入の扱いに注意 | ログ型API、CDC |
3) チェックポイント戦略
どこに「どこまで処理したか」を保存するかは重要です。選択肢を整理します。
| 方式 | 利点 | 欠点 |
|---|---|---|
| ローカルファイル（JSON/TXT） | シンプル・導入容易 | 複数実行環境では競合の危険 |
| DBメタテーブル | 排他や監査が可能 | DBのセットアップが必要 |
| オブジェクトストレージ（S3等） | 分散環境でも使える | アクセス遅延、コスト検討 |
4) アップサート／削除処理の実装パターン
代表的な実装例を短いコードで示します。依存は標準ライブラリ＋optionalでrequests／sqlite3のみ。
サンプル：CSV読み込み→差分計算→SQLiteへアップサート→チェックポイント保存
import csv
import hashlib
import json
import sqlite3
import argparse
from pathlib import Path
CHECKPOINT_FILE = 'checkpoint.json'
DB_FILE = 'target.db'
# 基本ユーティリティ
def read_csv(path):
with open(path, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
return list(reader)
def row_hash(row):
s = '|'.join(str(row.get(k,'')) for k in sorted(row.keys()))
return hashlib.md5(s.encode('utf-8')).hexdigest()
# チェックポイントの読み書き
def load_checkpoint():
p = Path(CHECKPOINT_FILE)
if not p.exists():
return {}
return json.loads(p.read_text(encoding='utf-8'))
def save_checkpoint(data):
Path(CHECKPOINT_FILE).write_text(json.dumps(data), encoding='utf-8')
# SQLiteアップサート（簡易例）
def ensure_table(conn):
conn.execute('''CREATE TABLE IF NOT EXISTS items (
id TEXT PRIMARY KEY,
payload TEXT,
row_hash TEXT
)''')
def upsert_row(conn, row_id, payload, rhash):
conn.execute('''INSERT INTO items(id,payload,row_hash)
VALUES(?,?,?)
ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, row_hash=excluded.row_hash
''', (row_id, json.dumps(payload), rhash))
# 差分計算と処理
def compute_and_apply(source_rows):
checkpoint = load_checkpoint()
prev_hashes = checkpoint.get('hashes', {})
new_hashes = {}
conn = sqlite3.connect(DB_FILE)
try:
ensure_table(conn)
for r in source_rows:
rid = r.get('id') # key列を想定
if not rid:
continue
h = row_hash(r)
new_hashes[rid] = h
if prev_hashes.get(rid) != h:
upsert_row(conn, rid, r, h)
conn.commit()
finally:
conn.close()
# 保存は最後に一度だけ
save_checkpoint({'hashes': new_hashes})
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('csvfile')
args = parser.parse_args()
rows = read_csv(args.csvfile)
compute_and_apply(rows)
上記は最小限の例です。実務ではトランザクション境界や例外時のロールバック、並行実行制御を追加してください。
5) 冪等性・トランザクション・ロールバックの考え方
重要なポイントは「何度実行しても問題ない」ことです。簡単な設計原則を示します。
- アップサートを基本にする（INSERT→UPDATEの組合せ）。
- チェックポイントは、処理が安全に完了した直後に更新する（処理途中で更新しない）。
- 外部APIの呼び出しは冪等トークンや条件付きPUTを使う。APIが対応していない場合は、実行前後で状態確認を行う。
- DBトランザクションは可能な限り短く、コミットは一括で行う。部分コミットがあると整合性エラーが起きやすい。
6) 障害時のリカバリ手順とテスト方法
実務で役立つ基本的なリカバリ手順と、テストの考え方を示します。
- まずはチェックポイントを確認してどの範囲が未完了かを把握する。
- 再実行はチェックポイント以降のみを処理する。チェックポイントが壊れている場合は、最後の安定状態からフルリカバリ（全件再同期）を検討する。
- テスト手順：1) 小さいテストCSVで正常終了を確認、2) 故意に途中で例外を発生させてチェックポイントの不整合を検証、3) 同期の再実行で整合性が取れることを確認する。
7) 性能・スケーリングの現実的な注意点
実務での落とし穴をまとめます。
| 問題 | 影響 | 対応例 |
|---|---|---|
| 大きなCSVをメモリで読み込む | メモリ不足・遅延 | ストリーミング読み／チャンク処理（第122回参照） |
| APIレート制限 | 呼び出し失敗・遅延 | バッチ化・バックオフ・キャッシュ |
| 並行実行による競合 | ダブルコミット・整合性破壊 | リーダーロック、DBでの排他、ジョブスケジューラ制御 |
8) 実務運用チェックリストとサンプルスクリプト配布案内
運用前に確認すべき項目をチェックリストにしました。導入前に一つずつ潰してください。
- チェックポイントの保存場所と権限を決めたか
- 失敗通知（メール／Slack）の仕組みを用意したか
- 再実行ポリシー（何回リトライするか）を決めたか
- ログの保存期間やサンプル保持（例：30日）を決めたか
- モニタリング指標（処理時間、処理件数、エラー率）を定義したか
運用パターン（簡易）
- cron／Windowsタスク：小規模で手軽。ログはファイル＋メール通知。
- GitHub Actions：コード管理とスケジュールを一元化。シークレット管理が楽。
- 本格的なオーケストレーション（Airflow等）：依存関係やリトライ、監視が必要な場合
失敗しやすいポイントと回避策（チェックリスト）
| 問題 | 原因 | 回避策 |
|---|---|---|
| タイムスタンプのずれ | サーバー間の時刻差 | UTCに統一、またはハッシュベースに切替 |
| 並行更新での競合 | 複数プロセスが同一レコードを処理 | DBの排他制御、ジョブ単位でロック |
| 部分コミットによる欠落 | チェックポイントを誤って早期更新 | チェックポイントは処理完了後に一括更新 |
| APIレート超過 | 短期間に多数リクエスト | バッファリング、バッチ呼び出し、バックオフ |
実行手順とコマンド例
上のサンプルスクリプトを使った実行手順の例です。
- 準備：Python（3.8+）を用意。外部依存は不要。API利用時はrequestsを追加。
- 実行例：
python sync_script.py source.csv - 定期実行：cron で毎朝4時に実行する例：
0 4 * * * /usr/bin/python3 /path/to/sync_script.py /path/to/source.csv >> /var/log/sync.log 2>&1 - GitHub Actions：ワークフローで schedule を使って定期実行可能
まとめ
差分同期は「どこまで処理したか（チェックポイント）」を確実に管理し、アップサートと冪等性を中心に設計すれば、小さな手間で大きな効率化が期待できます。本記事では、タイムスタンプ／ハッシュ／シーケンスIDといった差分検出の考え方、チェックポイントの置き方、簡易なPythonサンプル、運用時の注意点を提示しました。まずは小さなスクリプトで運用を回し、実際の障害を元に改善していくことをおすすめします。
関連回：第113回（ファイル入出力と例外処理）、第116回（設定・引数・ロギング）、第122回（大規模CSVのストリーミング）、第124回（再試行と冪等性）。次回は運用のワークフロー化（スケジューラ／オーケストレーション）に進みます。