初めに

サーバレスでイベント駆動な構成は便利です。しかし、サービスの仕様を把握せずに使うと思わない落とし穴にハマります。テーマは Exactly Once と At Least Once によるデータ重複です。
個々のネタは既出ですが、通しで記載したものは少ないということでご容赦ください。
サンプルコードは用意していません。GCP環境での事例ですが、イベント駆動におけるat least once問題は他のクラウド環境でも発生するので参考になる場面もあるでしょう。
間違いのある場合はコメントでツッコミお願いします。

よくある構成

以下のような構成を考えてみます。よく事例発表で見かける構成ですね。

起点はGCSへのファイル配置です。ファイルが配置されると、pubsubに対してイベントが発行されます。
次にpubsubからイベントがcloud functionsへ発行されます。cloud functionsはイベントからオブジェクト名を取り出し、該当するファイルの中身をBigQueryへ投入するものとします。なお話の都合上、cloud functionsにて何らかの加工処理を行い、pythonのBQ用ライブラリを使ってstreaming insertしているものとしましょう。pythonな理由は筆者が普段pythonしか触らないからです。

そんな構成で大丈夫か?

大丈夫だ、問題ない。ほんとにそうでしょうか。いくつかの公式ドキュメントを引いてみましょう。

GCS -> pubsub

GCSからpubsubへ通知を送る仕様ページに、delivery guaranteesの記載があります。
さて以下の1文が気になりますね。

Cloud Storage guarantees at-least-once delivery to Pub/Sub.

なるほど、GCSからpubsubへの通知は少なくとも1回であって、同じメッセージが複数回発行されることもあるのですね。ん??

pubsub -> cloud functions

ではpubsubからcloud functionsを起動するところの仕様を確認しましょう。
同じくpubsub-notificationsのdelivery_guaranteesから、先ほどの引用箇所の直後に以下の記述があります。

Pub/Sub also offers at-least-once delivery to the recipient

とても気になる記述ですね。pubsub -> cloud functionsの仕様をもう少し見てみましょう。
cloud functions側のexecution_guaranteesには以下の記載があります。

Event-driven functions are invoked at least once

Event-driven functionsとはなんぞやと思われるかもしれませんが、cloud functionsをpubsub triggerで構築している場合はevent-driven functionsのなようです。参考
つまり、pubsubはcloud functionsを単一メッセージに対して複数回起動する可能性があるということですね。おやおやー??

cloud functions -> BigQuery

冷や汗が出てきました。cloud functionsからBigQueryへ入れるときはどうでしょうか。pythonでstreaming insertを行う想定をします。また、ここでのstreaming insertは昔ながらのtabledata.insertAllで行う前提です。今回もドキュメントを読んでみましょう。

De-duplication offered by BigQuery is best effort, and it should not be relied upon as a mechanism to guarantee the absence of duplicates in your data.

はい。要約すると投入時にinsertIdを付与すると重複排除を頑張ってくれますが、特に保証があるというわけではないということです。何ということだ。

いつの間にかページタイトルにlegacyと付いたことに趣を感じます(フラグ)。

ここまでのまとめ

以下のタイミングでデータ重複が発生する可能性があると分かりました。

gcs -> pubsub
pubsub -> cloud functions
cloud functions -> BigQuery

全部ですね。何ということでしょう。

そんな構成で大丈夫か? again

一番いい構成を頼む。一番かは判りませんが、手の打ちやすい構成を考えてみましょう¹。

insetId無しの場合はstreaming insertを行う時点で無理な気がします。streaming insertを辞めて外部表からinsert selectする、またはGCSからロードの方式を取れば可能性は有ります。以下、insertId無しで考えてます²。

パターン1: 頑張ってexactly onceを実現する

必ずGCSのオブジェクト単位で1回しか取り込まれない様にする場合があります。この方法としてよく見かける解説はpubsubのevent_idやオブジェクト名をredis/RDB/firestoreで管理するというものですが、取り込み先のテーブルをオブジェクト名依存なものにしてテーブル作成の重複エラーで落とす方法もあります。ただしどちらにしても、落とした場合にpubsubへ正常で返信するのか異常で返信するのか悩ましいところです。なので筆者は基本的にこの方法は取りません。

パターン2: 重複を許容してしまう -- 力は全てを解決する

BigQueryクエリ性能に全幅の信頼を置き、演算能力で解決するパターンです。筆者は~~めんどくさいから~~こちらのパターンを取りがちです。
これを実現する方法は、レコードに重複排除用の識別キーを必ず持たせることです。キー値は上流のデータソースで保持していないデータでも構いません。例えばcloud functions中でレコードごとにハッシュ値を取る、でも大丈夫です。とりあえずハッシュ値ごとに1件を取り出せば何とかなるはず。ハッシュ値が被るパターンなどデータ要件によっては別途検討が要ります。
重複があったとしても重複排除に使える識別キーが付与されていればBQ内で対処することが可能です。

パターン3: BigQuery Storage Write API (未検証)

先日、storage write apiなるものがリリースされました。ドキュメントによるとexactly-once semanticsだそうです。クライアントライブラリは通常のbigquery用ではなく、storage api用に違うものがあるらしい。とても期待できそうなのですが筆者はまだ試していないのでご紹介のみに留めます。誰か記事書いてくれないかな・・・。

結論: イベント駆動は仕様を確認し適切に利用しましょう

イベント駆動は便利ですが、重複排除の観点で頭が痛いことが判りました。今回は重複の話しか考慮していませんので、本番用に道中での各種エラー対処なども加味するともう少し複雑な構成が必要になります。担当者のソフトウェアエンジニアリングが試されますね。
不安になった方は今すぐ本番環境を確認しましょう。読者の皆様の年末年始が穏かなものとなることを祈っております。

event arcは触ってないので触れません。pubsubのsubscriptionを自分で決めたい欲求が強くて・・・。 ↩
insertIdがあれば行けるのでは、と思いたいのですが1分の制限があるので厳しいのかなと考えています。いやでもそもそもこの重複排除機能自体がベストエフォートなんだよな・・・。 ↩