Salesforce Change Data Capture (CDC) を活用したリアルタイムデータ連携の実現

背景と適用シナリオ

現代のエンタープライズシステムアーキテクチャにおいて、システム間のデータ一貫性とリアルタイム性は、ビジネスの俊敏性と意思決定の速度を左右する極めて重要な要素です。Salesforce を中心的な顧客管理プラットフォームとして利用する企業が増える中、「Salesforce 上のデータ変更を、いかに迅速かつ効率的に他のシステムに伝播させるか」という課題は、私たち統合エンジニア (Integration Engineer) にとって常に主要なテーマとなります。

従来、この課題に対するアプローチとしては、定期的に Salesforce API をポーリングして変更を問い合わせるバッチ処理が一般的でした。しかし、この方法は API コール数を大量に消費し、リアルタイム性に欠け、変更がない場合でもリソースを無駄にするという欠点がありました。

ここで登場するのが Change Data Capture (CDC、変更データキャプチャ) です。CDC は、Salesforce 内のレコードに対する作成、更新、削除、復元といったデータ変更をイベントとしてリアルタイムに発行する、イベント駆動型アーキテクチャ (Event-Driven Architecture) を実現するための強力な機能です。私たち統合エンジニアは、この CDC を利用することで、効率的でスケーラブル、かつニアリアルタイムなデータ連携ソリューションを構築できます。

具体的な適用シナリオ

データウェアハウス (DWH) との同期: Salesforce の取引先 (Account) や商談 (Opportunity) などのオブジェクトデータを、分析目的で Snowflake、Google BigQuery、Amazon Redshift といった外部の DWH にニアリアルタイムで複製します。これにより、ビジネスインテリジェンス (BI) チームは常に最新のデータに基づいた分析を行うことができます。
外部システムとのデータ一貫性維持: 基幹業務システム (ERP) や自社開発のデータベースなど、外部システムに保持されている顧客マスタや商品マスタを Salesforce と同期させます。例えば、Salesforce で取引先責任者の連絡先が更新された際に、即座に ERP の顧客情報も更新するといった連携が可能です。
外部ワークフローのトリガー: Salesforce の商談フェーズが「成立」に更新されたことを CDC イベントとして捕捉し、これをトリガーとして外部の受注管理システムや請求システムで新たなプロセスを自動的に開始させます。
監査ログとコンプライアンス: 重要なデータの変更履歴を詳細に記録する必要がある場合、CDC イベントを永続的なストレージ（例: S3、ELK Stack）にアーカイブすることで、誰が、いつ、どのデータを変更したかの監査ログを構築できます。

原理説明

Change Data Capture の中核をなすのは、Publish/Subscribe (Pub/Sub) モデルです。Salesforce はイベントの発行者 (Publisher) となり、データ変更に関心のある外部システムは購読者 (Subscriber) となります。この仕組みは Salesforce の Platform Events (プラットフォームイベント) の堅牢なインフラストラクチャ上で構築されています。

イベント発行の仕組み

オブジェクトの有効化: Salesforce の管理者が、設定画面から CDC の対象としたいオブジェクト（標準オブジェクトまたはカスタムオブジェクト）を選択します。
トランザクションのコミット: ユーザーや自動化プロセスによって対象オブジェクトのレコードが作成、更新、削除、または復元されると、Salesforce はデータベーストランザクションをコミットします。
変更イベントの発行: トランザクションが正常にコミットされた後に、Salesforce はその変更内容を含む変更イベントメッセージを生成し、特定のイベントチャネルに発行します。これにより、データの整合性が保証されます。

変更イベントの構造

発行される変更イベントは、JSON 形式のメッセージであり、主に二つの要素で構成されています。

ChangeEventHeader: イベントのメタデータが含まれます。
- entityName: 変更が発生したオブジェクト名 (例: 'Account')。
- recordIds: 変更されたレコードの ID の配列。
- changeType: 変更の種類 ('CREATE', 'UPDATE', 'DELETE', 'UNDELETE')。
- transactionKey: 同じトランザクション内で発生したすべての変更イベントをグループ化するためのキー。
- commitTimestamp: トランザクションがコミットされたタイムスタンプ。
ペイロード: 変更されたレコードのフィールド値が含まれます。UPDATE の場合、変更があったフィールドのみが含まれるわけではなく、CDC が有効化されたオブジェクトのすべての非暗号化フィールドが含まれます。これにより、購読者側はレコードの最新状態を容易に把握できます。

イベントの購読

外部のクライアントアプリケーションは、CometD (CometDプロトコル) というプロトコルを使用して Salesforce のイベントバスに接続し、特定のチャネルを購読します。CometD は、HTTP ロングポーリングを利用してサーバーからのプッシュ通知をシミュレートする技術であり、効率的なリアルタイム通信を実現します。購読するチャネル名は、/data/<ObjectName>ChangeEvent という形式になります（例: /data/AccountChangeEvent）。

各イベントには ReplayId (リプレイID) と呼ばれる一意の識別子が割り当てられます。クライアントは、最後に正常に処理したイベントの ReplayId を保存しておくことで、ネットワーク切断などの障害から復旧した際に、見逃した可能性のあるイベントを Salesforce に再要求し、メッセージの欠損を防ぐことができます。Salesforce はイベントを最大72時間保持します。

示例代码

ここでは、Java を使用して Salesforce の取引先オブジェクト (Account) の変更イベントを購読するクライアントの実装例を示します。この例では、Salesforce が提供する公式ライブラリである EMP Connector を使用します。EMP Connector は、CometD 接続の複雑な詳細（認証、ハンドシェイク、再接続ロジックなど）を抽象化してくれる便利なツールです。

このコードは、指定されたチャネル（この場合は /data/AccountChangeEvent）を購読し、イベントを受信するたびにその内容をコンソールに出力します。

import java.net.URL;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.cometd.bayeux.Channel;
import org.cometd.bayeux.Message;
import org.cometd.client.BayeuxClient;
import org.cometd.client.transport.LongPollingTransport;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.salesforce.emp.connector.BayeuxParameters;
import org.salesforce.emp.connector.EmpConnector;
import org.salesforce.emp.connector.LoginHelper;
import org.salesforce.emp.connector.TopicSubscription;

public class ChangeDataCaptureClient {

    public static void main(String[] argv) throws Exception {
        // Salesforce へのログイン情報と接続パラメータ
        // 本番環境では、これらの値をハードコーディングするのではなく、
        // 設定ファイルや環境変数から安全に読み込むべきです。
        String username = "your_username@example.com";
        String password = "your_password_and_security_token";
        String loginUrl = "https://login.salesforce.com"; // Sandbox の場合は https://test.salesforce.com

        // BayeuxParameters を使用して Salesforce への接続情報を設定
        // LoginHelper.login() は OAuth 2.0 パスワードフローを使用してアクセストークンを取得します。
        BayeuxParameters parameters = LoginHelper.login(new URL(loginUrl), username, password);

        // Consumer を定義して、受信したイベントを処理するロジックを実装
        // この例では、受信したイベントデータを単純にコンソールに出力します。
        Consumer<Map<String, Object>> consumer = event -> System.out.println(String.format("Received event: %s", event));

        // EmpConnector のインスタンスを作成
        EmpConnector empConnector = new EmpConnector(parameters);

        // CometD のエラーハンドリング
        empConnector.setConnectionListener(() -> System.out.println("Connected to Salesforce."));
        empConnector.setFailureListener((t) -> {
            System.err.println("CometD connection failed.");
            t.printStackTrace(System.err);
        });

        // EmpConnector の実行を開始
        // これにより、Salesforce への認証と CometD ハンドシェイクが行われます。
        empConnector.start().get(5, TimeUnit.SECONDS);

        // 購読したい CDC チャネル名
        // '/data/AccountChangeEvent' は取引先オブジェクトの変更イベントチャネルです。
        String channel = "/data/AccountChangeEvent";

        // 指定したチャネルを購読
        // ReplayId.LATEST を指定すると、購読開始後に発生した新しいイベントのみを受信します。
        // 特定の ReplayId を指定して、過去のイベントから再生することも可能です。
        long replayId = TopicSubscription.REPLAY_FROM_LATEST; 
        TopicSubscription subscription = empConnector.subscribe(channel, replayId, consumer).get(5, TimeUnit.SECONDS);

        System.out.println(String.format("Subscribed to channel: %s", subscription.getTopic()));
        System.out.println("Waiting for events... Press a key to exit.");

        // クライアントを稼働させ続けるための待機処理
        // 実際のアプリケーションでは、デーモンスレッドなどで実行されます。
        System.in.read();

        // 終了時に EmpConnector を停止
        empConnector.stop();
    }
}

注意事項

Change Data Capture を本番環境で運用する際には、いくつかの重要な点に注意する必要があります。

権限設定

ユーザ権限: イベントを購読する API ユーザには、適切なプロファイルまたは権限セットが必要です。最低でも「すべてのデータの参照」権限と、システム権限の「変更データキャプチャの変更を受信」が必要です。
オブジェクト権限: ユーザは、CDC が有効になっているオブジェクトに対する参照アクセス権を持っている必要があります。

API 制限と割り当て

イベント配信の割り当て: Salesforce 組織には、24時間以内に配信できるイベント数の上限があります。この上限は Salesforce のエディションによって異なります。上限に達すると、新たなイベントの配信が一時停止されるため、[設定] > [組織情報] で使用状況を定期的に監視することが不可欠です。
有効化できるエンティティ数: CDC を有効化できるオブジェクトの数にも上限があります。大規模な組織では、どのオブジェクトを対象にするか慎重に選択する必要があります。

データ保持と ReplayId

72時間の保持期間: 変更イベントはイベントバスに最大72時間（3日間）保持されます。クライアントアプリケーションが3日以上停止していた場合、停止期間中に発生したイベントは失われる可能性があります。
永続的な ReplayId の保存: 堅牢な統合を構築するためには、クライアント側で最後に処理した ReplayId をデータベースやファイルシステムなどの永続的なストレージに保存することが必須です。クライアントが再起動する際には、保存された ReplayId を使用して購読を再開し、イベントの欠損や重複処理を防ぎます。

エラーハンドリングとクライアントの堅牢性

接続障害: ネットワークの問題や Salesforce 側のメンテナンスにより、CometD 接続が切断される可能性があります。EMP Connector のようなライブラリは自動再接続を試みますが、クライアントアプリケーションはこれらのシナリオを適切に処理できるように設計する必要があります。
メッセージ処理のべき等性 (Idempotency): イベントは「少なくとも1回 (at-least-once)」配信される保証モデルです。つまり、稀に同じイベントが複数回配信される可能性があります。したがって、購読者側の処理ロジックは、同じイベントを複数回処理しても問題が発生しないように、べき等に設計する必要があります。

まとめとベストプラクティス

Salesforce Change Data Capture は、私たち統合エンジニアがリアルタイムデータ連携を構築するための、非常に強力でスケーラブルなツールです。API ポーリングに代わるイベント駆動型のアプローチを採用することで、API コール数を削減し、システム全体のパフォーマンスを向上させ、ビジネス要件に即した迅速なデータ連携を実現できます。

ベストプラクティス

適切なユースケースの選択: レコードデータの変更を忠実に複製・同期するシナリオに CDC を使用します。より複雑なビジネスプロセスを伝達したり、カスタムペイロードを送信したりする必要がある場合は、カスタムの Platform Events の方が適している場合があります。
堅牢なクライアントの実装: EMP Connector のような実績のあるライブラリを利用し、ReplayId の永続化、べき等なメッセージ処理、接続障害からの自動回復など、本番運用に耐えうるクライアントを構築してください。
API 制限の監視: 本番稼働後は、Salesforce のイベント配信割り当てを定期的に監視し、必要に応じて Salesforce に上限緩和を申請するか、アーキテクチャの見直しを検討します。
セキュリティの考慮: 接続ユーザの権限を最小権限の原則に従って設定し、パスワードやアクセストークンなどの認証情報は安全に管理してください。
十分なテスト: 大量のデータ変更が発生した場合のパフォーマンスや、クライアントが長時間停止した後のリカバリシナリオなど、様々な状況を想定したテストを徹底的に行い、システムの安定性を確認してください。

Change Data Capture を正しく理解し、これらのベストプラクティスを遵守することで、Salesforce を中心としたエンタープライズエコシステム全体の価値を最大化する、信頼性の高い統合ソリューションを設計・構築することが可能になります。

にゃんこのSalesforce日記

このブログを検索