Build a CDC application with EventBridge

はじめに

CDC (Change Data Capture) とは、上流のデータ変更を監視し、その変更情報を下流のビジネスに同期してさらに処理を行うアプリケーションシナリオを指します。近年、イベント駆動アーキテクチャ (EDA) の人気度が徐々に高まり、プロジェクトのアーキテクチャ設計者の第一選択肢となっています。EDA は、データの変更をイベントとして扱う CDC の基盤インフラと自然に適合します。各サービスは、自身に関心のあるイベントをリッスンすることで一連のビジネスドライバーを完了します。Alibaba Cloud EventBridge は、Alibaba Cloud が提供するサーバーレスイベントバスサービスであり、ユーザーが EDA アーキテクチャに基づいたアプリケーションを簡単に、かつ迅速に構築できるよう支援します。このたび、EventBridge イベントストリームは Alibaba Cloud DTS [1] サービスをベースにした CDC 機能をサポートするようになりました。この記事では、CDC の概要、EventBridge での CDC アプリケーション、およびいくつかのベストプラクティスシナリオから、EventBridge を使用して簡単に CDC アプリケーションを構築する方法を紹介します。

CDC の概要

基本原理とアプリケーションシナリオ

CDC はソースデータベースから増分データおよびデータスキーマの変更をキャプチャし、高い信頼性と低レイテンシのデータ転送を通じて、これらの変更をターゲットデータベース、データレイク、またはその他のデータ分析サービスに順序立てて同期します。現在、業界で主流のオープンソース CDC ツールには、Debezium [2]、Canal [3]、Maxwell [4] があります。

画像出典: https://dbconvert.com

現在、業界には主に以下のタイプの CDC 実装方式があります。

1. タイムスタンプまたはバージョン番号ベース

タイムスタンプベースの方式では、データベーステーブル内に更新タイムスタンプを表すフィールドが必要です。データの挿入や更新が行われると、対応するタイムスタンプフィールドもそれに応じて更新されます。CDC コンポーネントは、更新時刻が前回同期時刻より大きいデータレコードを定期的に取得し、この周期内のデータ変更をキャプチャします。バージョン番号ベースの追跡とタイムスタンプベースの追跡の原理は基本的に一致しており、開発者がデータを変更する際にデータのバージョン番号を更新する必要があります。

2. スナップショットベース

スナップショットベースの CDC 実装では、ストレージレベルでデータソースの 3 つのコピー、すなわち元のデータ、以前のスナップショット、現在のスナップショットを使用します。2 つのスナップショット間の差分を比較することで、データの変更内容を取得します。

3. トリガーベース

トリガーベースの CDC 実装は、ソーステーブルにトリガーを作成し、データ変更操作 (INSERT、UPDATE、DELETE) のレコードを保存することです。たとえば、ユーザーの変更を記録する専用テーブルを作成し、INSERT、UPDATE、DELETE の 3 種類のトリガーを作成して、ユーザーの変更をこのテーブルに同期します。

4. ログベース

前述の 3 つの方式はソースデータベースに対してある程度の侵入性がありますが、ログベースの方式は非侵入型の CDC 手法です。データベースはトランザクションログを使用してディザスタリカバリを実現します。たとえば、MySQL の binlog には、ユーザーによるデータベースへのすべての変更操作が記録されます。ログベース CDC では、トランザクションログを継続的に監視することで、データベースの変更をリアルタイムで取得します。

CDC の応用範囲は広く、リモートデータセンター間のデータベース同期、異種データベース間のデータ同期、マイクロサービスのデカップリング、キャッシュ更新、CQRS など、さまざまな側面を含みますが、これらに限定されません。

Alibaba Cloud ベースの CDC ソリューション: DTS

DTS (Data Transmission Service) は、Alibaba Cloud が提供するリアルタイムデータフローサービスです。リレーショナルデータベース (RDBMS)、非リレーショナルデータベース (NoSQL)、多次元データ分析 (OLAP) などのデータソース間のデータ連携をサポートし、データ同期、移行、サブスクリプション、統合、処理を一体化しています。DTS のデータサブスクリプション [5] 機能は、自社構築 MySQL や RDS MySQL などのデータベースからリアルタイムで増分データを取得するのに役立ちます。

EventBridge での CDC の応用

Alibaba Cloud EventBridge は、イベントバス [6] とイベントストリーム [7] という 2 つの異なるアプリケーションシナリオ向けのイベントルーティングサービスを提供します。

イベントバスの基盤レイヤーにはイベントの永続化機能があり、必要に応じてイベントを複数のイベントターゲットにルーティングできます。

イベントストリームはエンドツーエンドのストリーミングデータ処理シナリオに適しています。ソース側で生成されたイベントをリアルタイムで抽出、変換、分析、ロードしてターゲット側に送信し、イベントバスを作成する必要がないため、エンドツーエンドのダンプがより効率的で使いやすいです。

CDC シナリオでのユーザーのニーズにさらに対応するため、EventBridge はイベントストリームのソース側で Alibaba Cloud DTS のデータサブスクリプション機能をサポートしています。ユーザーはシンプルな設定でデータベースの変更を EventBridge イベントストリームに同期できます。

EventBridge は DTS SDK ベースの DTS Source Connector をカスタマイズしました。ユーザーが DTS をイベントプロバイダーとしてイベントストリームを設定すると、ソースコネクタは DTS サーバーから DTS レコードデータをリアルタイムでプルします。データがローカルにプルされた後、ID、operationType、topicPartition、beforeImage、afterImage などのデータを保持しつつ、一定の構造的カプセル化が行われ、ストリーミングイベントに必要なシステム属性が追加されます。

DTS イベントサンプルの詳細は、EventBridge の公式ドキュメントを参照してください。

「EventBridge Streaming は DTS イベントの順序を保証しますが、イベントが重複配信される可能性があります。EventId は各 DTS レコードと 1 対 1 のマッピング関係を確保しているため、ユーザーはこのフィールドに基づいてイベントの冪等処理を行うことができます。」

DTS ソースの EventBridge イベントストリームを作成する

以下では、EventBridge コンソールで DTS ソースのイベントストリームを作成する方法を示します。

- 事前準備



1. EventBridge サービスを有効化する。

2. DTS データサブスクリプションタスクを作成する。

3. サブスクリプションデータを消費するためのコンシューマーグループアカウント情報を作成する。

- イベントストリームの作成



1. EventBridge コンソールにログインし、左側のナビゲーションバーをクリックして「イベントフロー」を選択し、イベントフローリストページで「イベントフローの作成」をクリックします。

「基本情報」の「イベントフロー名」と「説明」は必要に応じて入力できます。

イベントストリームを作成してイベントプロバイダーを選択する際、ドロップダウンボックスから「データベース DTS」を選択します。

4. 「データサブスクリプションタスク」列で作成した DTS データサブスクリプションタスクを選択します。コンシューマーグループ列で、サブスクリプションデータの消費に使用するコンシューマーグループを選択し、コンシューマーグループのパスワードと初期消費時間を入力します。

5. イベントフロールールとターゲットを必要に応じて入力し、保存して開始すると、DTS データサブスクリプションをイベントソースとするイベントストリームが作成されます。

注意事項

使用する際は、以下の点に注意する必要があります。

「EventBridge は SUBSCRIBE 消費モード [8] を使用するため、現在の DTS コンシューマーグループ内で他のクライアントインスタンスが実行されていないことを確認してください。」設定したコンシューマーグループが以前に実行されていた場合、渡されたポジションは無効化され、そのコンシューマーグループが以前に消費していたポジションに基づいて消費が継続されます。

DTS イベントソース作成時に渡されたポジションは、新しいコンシューマーグループが初めて実行されるときにのみ有効になります。その後、タスクが再起動されると、最後の消費ポジションに基づいて消費が継続されます。

3. EventBridge イベントストリームは、OperationType が INSERT、DELETE、UPDATE、DDL の DTS データをサブスクライブします。

4. DTS イベントソースを使用すると、メッセージが重複する可能性があります。つまり、メッセージの損失は保証されませんが、1 回だけの配信は保証できません。ユーザーは冪等処理を行うことを推奨します。

5. 順序消費を確保する必要がある場合、例外許容ポリシーを「NONE」に設定する必要があります。これは、例外を許容しないことを意味します。この場合、イベントフローのターゲット側で異常なメッセージを消費すると、ターゲット側が正常に戻るまでイベントフロー全体が一時停止します。

ベストプラクティスの例

EventBridge を使用した CQRS の実装

CQRS (Command Query Responsibility Segregation) モデルでは、コマンドモデルが書き込みと更新操作を実行し、クエリモデルが効率的な読み取り操作をサポートします。読み取りと書き込みに使用するデータモデルには一定の差異があり、データ同期を確保するための一定の方法が必要です。EventBridge イベントストリームベースの CDC は、このようなニーズを満たすことができます。

クラウドサービスを使用することで、ユーザーは以下の方法で EventBridge ベースの CQRS を簡単に構築できます。

1. コマンドモデルでデータベースの変更操作を行い、クエリモデルで Elasticsearch (ES) からデータを読み取ります。

2. DTS データサブスクリプションタスクを有効化して DB の変更をキャプチャします。

3. EventBridge イベントフローを設定します。イベントプロバイダーは DTS データサブスクリプションタスク、イベントレシーバーは Function Compute (FC) です。

4. FC 内のサービスが Elasticsearch データの更新操作を行います。

マイクロサービスのデカップリング

CDC はマイクロサービスのデカップリングにも使用できます。たとえば、以下は E コマースプラットフォームの注文処理システムです。新しい未払い注文が生成されると、データベースに INSERT 操作が行われ、注文のステータスが「未払い」から「支払い済み」に変化すると、データベースに UPDATE 操作が行われます。注文ステータスの変化に応じて、バックエンドで異なるマイクロサービスが対応する処理を行います。

1. ユーザーが注文/支払いを行うと、注文システムがビジネス処理を実行し、データの変更を DB に書き込みます。

2. 新しい DTS サブスクリプションタスクを作成して DB データの変更をキャプチャします。

3. EventBridge イベントストリームを構築します。イベントプロバイダーは DTS データサブスクリプションタスク、イベントレシーバーは RocketMQ です。

RocketMQ データの消費時に、同じトピックの下に 3 つのグループを有効化して、異なるビジネス消費ロジックを表します。

A. GroupA は、キャプチャした DB 変更のユーザーキャッシュを更新し、ユーザーが注文ステータスをクエリしやすくします。

B. GroupB の下流は財務システムと連携しており、新規注文、つまり DB 操作タイプが INSERT のイベントのみを処理し、他のタイプのイベントは破棄します。

C. GroupC は、注文ステータスが「未払い」から「支払い済み」に変化したイベントのみに注目します。条件を満たすイベントが到着すると、下流の物流・倉庫システムを呼び出して注文をさらに処理します。

インターフェイス呼び出し方式を使用した場合、ユーザーが注文後に注文システムがキャッシュ更新インターフェイス、新規注文インターフェイス、注文支払いインターフェイスを呼び出す必要があり、ビジネスの結合度が高くなります。さらに、このモードでは、データコンシューマーは上流の注文処理インターフェイスが返す内容のセマンティック情報を気にする必要がありません。ストレージモデルが変わらなければ、データレベルで直接、このデータ変更を処理するかどうか、およびどのように処理するかを判断できます。同時に、メッセージキューの天然のメッセージ蓄積能力により、注文ピーク時のピークカットと谷埋めも実現できます。

実際、現在 EventBridge Streaming がサポートするメッセージングプロダクトには、RocketMQ の他に RabbitMQ、ApsaraMQ for Kafka、Message Service (MNS) なども含まれます。実際の運用では、ユーザーは必要に応じて選択できます。

データベースバックアップと異種データベースの同期

データベースのディザスタリカバリと異種データベース間のデータ同期も、CDC の重要なアプリケーションシナリオです。Alibaba Cloud EventBridge を使用して、こうしたアプリケーションも迅速に構築できます。

1. 新しい DTS データサブスクリプションタスクを作成して、ユーザーの MySQL データベースの変更をキャプチャします。

2. EventBridge イベントストリームを構築します。イベントプロバイダーは DTS データサブスクリプションタスクです。

3. EventBridge を使用してターゲットデータベースで指定された SQL を実行し、データベースバックアップを実現します。

4. データ変更イベントを Function Compute (FC) に配信し、ユーザービジネスがデータ変更内容に基づいて対応する異種データベースを更新します。

SQL 監査の自社構築

SQL 監査の自社構築が必要なユーザーも、EventBridge を使用すれば簡単に実装できます。

1. 新しい DTS データサブスクリプションタスクを作成してデータベースの変更をキャプチャします。

2. DTS をイベントプロバイダー、Simple Log Service (SLS) をイベントレシーバーとして EventBridge イベントストリームを構築します。

3. SQL 監査が必要な場合、SLS でクエリを実行できます。

まとめ

この記事では、CDC の概念、EventBridge での応用、およびいくつかのベストプラクティスシナリオを紹介しました。サポートするプロダクトの増加に伴い、EventBridge が支えるエコシステムの領域も拡大し続けています。メッセージエコシステムからデータベースエコシステムへ、ログエコシステムからビッグデータエコシステムへ、EventBridge は応用分野を広げ続け、クラウド上のイベントハブとしての地位を強固なものにしています。これからもこの方向で発展を続け、より深い技術とより幅広いエコシステムを実現していきます。