ETL差分転送の落とし穴 - Django ORM利用時の注意点と対策

この記事はニーリーアドベントカレンダー10日目です。

こんにちは！Analyticsチームの上田です。
データ分析基盤を運用していると、「データの鮮度を上げたい」「転送コストを下げたい」という欲求は尽きないものです。ちょうど私たちもETLの改善に取り組んでいたのですが、そこには思わぬ落とし穴が潜んでいました。。🥲
今年の反省は今年のうちにということで、得た学びを共有したいと思います！🙏

3行まとめ

データ転送コスト削減のため転送方式を 全件洗替え転送 → 差分転送 に切り替えたところ、転送漏れが発生
原因は、Django ORMのsave(update_fields=...) 利用時に、更新日時 (updated_datetime) が自動更新されない仕様を見落としていたこと
現在は洗替えに切り戻し、再発防止策としてLinterによるコード検知の仕組みを導入。併せてCDC (変更データキャプチャ) の導入も検討中

背景

ニーリーのデータ基盤では、プロダクトのデータベース (PostgreSQL) からGoogle BigQueryへのデータ転送にETLツールのTROCCOを利用しています。
これまで、日次のデータ転送は全件洗替えで実施していましたが、サービス成長・分析の拡大に伴いデータ量が増加し、転送にかかる時間やコストが無視できない規模になってきました。

そこで私たちは、より効率的な差分更新への切り替えを実施しました。具体的には、テーブル内の updated_datetime (更新日時) カラムを参照し、前回転送以降に更新されたレコードだけをBigQueryに送る方式です。これにより、転送時間を大幅に短縮できるはずでした。

※ この方式ではdeleteは反映できないため、deleteが行われないテーブルに限定して実施しました。

問題

転送方式を切り替えてから数日後、異変が発覚します。特定条件のユーザー抽出を行うクエリの結果に、本来含まれるべきユーザーが含まれていないことが判明しました。
調査の結果、「プロダクト側でレコードは更新されているのに、BigQuery側にその変更が反映されていない」ケースがあることがわかりました。差分転送の要である updated_datetime が、特定の更新処理において更新されていませんでした。

幸い、切替前の転送設定を残していたため、切り戻しは迅速に実施できました。また、日次のデータもバックアップが取られていたため、復元することができました。

原因

技術面

Park DirectのバックエンドではフレームワークとしてDjangoを利用していますが、そのORMの仕様に関する認識不足がありました。具体的には、DjangoのModel (私たちが利用しているバージョン) は、 save(update_fields=['status']) のように更新するフィールドを限定して保存した場合、updated_datetime を明示的に update_fields に含めない限り、自動更新されないという仕様があったのです。

結果として、この特定の処理を通ったレコードは「更新されたのに更新日時が古いまま」となり、ETLツールの差分検知ロジックから漏れてしまいました。

※ 私としては上記Django Modelの仕様に落ち度は無いと思っています🙏

プロセス面

差分転送に切り替えた後の転送漏れのチェックが不十分でした。切替前 (全件洗替え転送) はTROCCOのアラート機能で転送漏れをチェックできていたのですが、切替後は前提条件が変わったため、転送のスキップ行の有無しかチェックできていませんでした。結果として、データ基盤側で転送漏れを自己検知できませんでした。

また、差分転送の事前検証ではデータ転送漏れの有無も検証していたのですが、検証対象のテーブルを一部に限定していたため、それらのテーブルでは update_fields が利用されておらず、問題を検知できませんでした。

再発防止策

静的解析によるガード (Linterの導入)
ARCHチームにより、Djangoのコード内で save() メソッド等を使用する際に更新日時の指定漏れがないかを検知するflake8 pluginが作成・導入されました。人間が気をつけるのではなく、仕組みで検知するアプローチです。
CDC (Change Data Capture) の導入検討
updated_datetime というアプリケーションロジックに依存する差分転送には他にも弱点が見えてきたため、データベースのトランザクションログから直接変更を検知するCDCへの移行検証を開始しました。
プロセス改善
転送方式見直し時のルールに 新旧方式の並行稼働期間を設け、アラートによる整合性チェックを行うこと を盛り込みました。基盤側で事前に自己検知できるようにするためです。