Salesforce Change Data Capture を極める：開発者のためのリアルタイム統合ガイド

概要とビジネスシーン

Salesforce Change Data Capture (CDC) は、Salesforce オブジェクトのデータ変更をほぼリアルタイムでキャプチャし、イベントストリームとして外部システムや他の Salesforce サービスに配信するための強力な機能です。これにより、データ同期の遅延を最小限に抑え、システム間の疎結合な統合を可能にし、ビジネスプロセスの応答性を大幅に向上させます。

実際のビジネスシーン

シーンA - 小売業界：あるECサイト運営企業では、Salesforce の顧客データを基幹システムやマーケティングオートメーションツールとリアルタイムで同期する必要がありました。従来のバッチ処理では、データ更新に数時間の遅延が発生し、顧客セグメンテーションやプロモーションのタイムリーな実施が困難でした。
ソリューション：CDC を利用して Salesforce の取引先や商談オブジェクトの変更イベントをリアルタイムでキャプチャし、外部のデータウェアハウスにストリーミング。
定量的効果：顧客データの同期遅延を「数時間」から「数秒」に短縮し、マーケティングキャンペーンの即時性を高め、顧客エンゲージメントが15%向上しました。

シーンB - 金融業界：複数の銀行システムをSalesforceと連携させている金融機関では、顧客の口座情報や取引履歴の変更を即座に検知し、不正取引監視システムや顧客サポートシステムに通知する必要がありました。複雑なポーリング処理はシステム負荷が高く、リアルタイム性に欠けていました。
ソリューション：CDC を Salesforce のカスタムオブジェクトに適用し、関連する変更イベントをリアルタイムで不正検知システムにプッシュ。
定量的効果：不正検知までの時間を大幅に短縮し、年間で数百万ドルの潜在的な損失を防ぐことに貢献。顧客サポート担当者も最新の顧客情報に即座にアクセスできるようになり、顧客満足度が向上しました。

シーンC - 製造業：製造業の企業が生産ラインのIoTセンサーデータや在庫情報をSalesforceのカスタムオブジェクトで管理しています。生産ラインでの部品在庫の変更をリアルタイムでERPシステムに連携し、サプライチェーンの最適化を図りたいと考えていました。
ソリューション：Salesforce の在庫オブジェクトに CDC を適用し、在庫レベルの変更イベントをERPシステムにリアルタイムで通知。
定量的効果：在庫切れによる生産ライン停止のリスクを50%削減し、Just-In-Time (JIT) 生産戦略の精度を向上させました。これにより、年間運用コストを3%削減することに成功しました。

技術原理とアーキテクチャ

Change Data Capture は、Salesforce データベースにおける標準オブジェクトおよびカスタムオブジェクトの挿入（Insert）、更新（Update）、削除（Delete）、復元（Undelete）といったデータ操作イベントを自動的にキャプチャし、変更イベント（Change Event）として Salesforce イベントバス（Event Bus）に発行します。これらの変更イベントは、Publish-Subscribe（出版-購読）モデルに基づいて動作し、リアルタイムでのデータ統合を可能にします。

主要コンポーネントと依存関係

変更イベント (Change Event)：Salesforce オブジェクトのデータ変更を表す特別な種類のプラットフォームイベント（Platform Event）。各変更イベントには、ChangeEventHeader と呼ばれるメタデータが含まれており、変更の種類、変更されたフィールド、レコードID、トランザクション情報などが提供されます。
Salesforce イベントバス (Event Bus)：変更イベントを含むすべてのプラットフォームイベントをルーティングし、購読者に配信する高性能なメッセージキューイングシステムです。
Streaming API (CometD)：外部システムが変更イベントをリアルタイムで購読するための API です。CometD プロトコルを利用して、HTTP ロングポーリングを通じてイベントをクライアントにプッシュします。
Platform Event Trigger：Salesforce 組織内で変更イベントを購読し、Apex コードを非同期で実行するためのトリガーです。通常のオブジェクトトリガーと同様に動作しますが、イベントオブジェクトに対して定義されます。

データフロー

ステップ	説明	関連コンポーネント
1. データ変更	Salesforce オブジェクトに対して DML (Insert, Update, Delete, Undelete) 操作が実行されます。	Salesforce Database
2. イベントキャプチャ	Salesforce システムが自動的にデータ変更を検出し、変更イベントを生成します。	Change Data Capture サービス
3. イベント発行	生成された変更イベントが Salesforce イベントバスに発行されます。	Event Bus
4. イベント購読 (外部)	外部システムは Streaming API (CometD クライアント) を介してイベントバスから変更イベントをリアルタイムで購読します。	Streaming API, CometD クライアント
5. イベント購読 (内部)	Salesforce 組織内の Platform Event Trigger がイベントバスから変更イベントを購読し、Apex コードを実行します。	Platform Event Trigger, Apex
6. 処理と連携	購読したシステムや Apex がイベントデータを処理し、必要なビジネスロジックを実行したり、他のシステムと連携したりします。	外部システムロジック, Apex Business Logic

ソリューション比較と選定

ソリューション	適用シーン	パフォーマンス	Governor Limits	複雑度
Change Data Capture	リアルタイムに近いデータ同期、疎結合なシステム連携、監査ログ、ストリーミング分析	高 (ミリ秒単位の遅延)	イベント発行/購読の制限、ストリーミング接続制限	中 (設定と購読クライアント実装)
Apex Trigger + Callout/Queueable	同期的な外部システム連携、複雑なビジネスロジックの即時実行	中～高 (Callout の場合はネットワーク遅延に依存)	Callout 制限、CPU時間、DML制限など厳格	高 (トリガー、非同期 Apex、エラーハンドリング)
Polling (API Batch Polling)	定期的なデータ同期、バッチ処理、リアルタイム性が不要な場合	低 (ポーリング間隔に依存)	API コール数制限、クエリ制限	低～中 (API クライアント実装)

change data capture を使用すべき場合

✅ **リアルタイムに近いデータ同期が必要な場合**：データ変更を即座に検知し、外部システムや他の Salesforce アプリケーションに反映させたい場合。
✅ **疎結合なシステム統合を目指す場合**：Salesforce と外部システム間の依存関係を低減し、耐障害性の高いアーキテクチャを構築したい場合。
✅ **高スループットなイベントストリーム処理が必要な場合**：大量のデータ変更イベントを効率的に処理し、スケーラビリティを確保したい場合。
✅ **データ変更の監査や履歴管理を行いたい場合**：どのデータが、いつ、どのように変更されたかを記録し、分析したい場合。

❌ **不適用シーン**：

Salesforce 内で同期的なデータ検証や計算が必要な場合（この場合はApex Triggerが適しています）。
ごく少量のデータを数分から数時間おきに同期すれば十分な場合（API Batch Pollingでも十分な場合があります）。

実装例

ここでは、Salesforce オブジェクトの変更イベントを有効化し、Salesforce 内部で Platform Event Trigger を使って購読する例と、外部システムから CometD を使って購読する際の概念的なコード例を示します。

1. Change Data Capture の有効化

これはコードではなく、Salesforce の設定画面から行います。

Salesforce の「設定 (Setup)」に移動します。
クイック検索ボックスで「変更データキャプチャ (Change Data Capture)」と入力し、選択します。
利用可能なエンティティのリストから、変更イベントをキャプチャしたいオブジェクト（例: Account、CustomObject__c）を選択し、「選択済みのエンティティ (Selected Entities)」リストに移動します。
「保存 (Save)」をクリックします。

これで、選択したオブジェクトに対する DML 操作が発生すると、自動的に変更イベントが発行されるようになります。

2. Salesforce 内部での購読例 (Platform Event Trigger)

ここでは、Account オブジェクトの変更イベントを購読し、変更内容をデバッグログに出力する Apex Platform Event Trigger の例です。

// AccountChangeEvent トリガー
// Account の変更イベントが発生した際に実行されます。
trigger AccountChangeEventTrigger on AccountChangeEvent (after insert) {
    // 変更イベントのリストをループ処理します。
    for (AccountChangeEvent event : Trigger.New) {
        // ChangeEventHeader を取得し、変更の種類やフィールド情報を確認します。
        // ChangeEventHeader は変更イベントのメタデータを提供します。
        ChangeEventHeader header = event.ChangeEventHeader;
        System.debug('--- ChangeEventHeader ---');
        System.debug('EntityName: ' + header.entityName); // 変更されたオブジェクト名 (例: Account)
        System.debug('ChangeType: ' + header.changeType); // 変更の種類 (例: UPDATE, CREATE, DELETE)
        System.debug('RecordIds: ' + header.recordIds); // 変更されたレコードの ID リスト
        System.debug('ChangedFields: ' + header.changedFields); // 変更されたフィールドのリスト

        // 変更イベントのペイロードをデバッグ出力します。
        // AccountChangeEvent オブジェクト自体が変更後の Account データを保持します。
        System.debug('--- AccountChangeEvent Payload ---');
        System.debug('Name: ' + event.Name);
        System.debug('Industry: ' + event.Industry);
        System.debug('Phone: ' + event.Phone);

        // 変更の種類に基づいて特定の処理を行う例
        if (header.changeType == 'UPDATE') {
            // ここに更新時のカスタムロジックを記述
            System.debug('Account ' + event.Name + ' (ID: ' + header.recordIds[0] + ') が更新されました。');
        } else if (header.changeType == 'CREATE') {
            // ここに作成時のカスタムロジックを記述
            System.debug('Account ' + event.Name + ' (ID: ' + header.recordIds[0] + ') が作成されました。');
        }
        // 他の変更タイプ (DELETE, UNDELETE) も同様に処理可能
    }
}

実装ロジック解析:

trigger AccountChangeEventTrigger on AccountChangeEvent (after insert): この行は、AccountChangeEvent という名前のイベントオブジェクトに対してトリガーを定義しています。(after insert) は、新しい変更イベントがイベントバスに発行されたときにトリガーが実行されることを意味します。
for (AccountChangeEvent event : Trigger.New): トリガーコンテキストの Trigger.New には、現在のトランザクションで発行されたすべての変更イベントが含まれています。各イベントをループ処理します。
ChangeEventHeader header = event.ChangeEventHeader;: 各変更イベントは ChangeEventHeader という特別なプロパティを持っており、これにより変更の種類 (changeType)、変更されたエンティティ名 (entityName)、影響を受けたレコードの ID (recordIds)、変更されたフィールド (changedFields) などのメタデータにアクセスできます。
System.debug(...): 変更イベントのペイロード (例: event.Name, event.Industry) とヘッダー情報をログに出力しています。実際のビジネスロジックでは、これらの情報を使って後続の処理（例: 関連レコードの更新、外部システムへのコールアウト）を実行します。

3. 外部システムでの購読例 (CometD クライアント - 概念)

外部システムは、Streaming API を介して変更イベントを購読します。ここでは、Java を使用した CometD クライアントの基本的な構造を示しますが、実際のライブラリはより詳細な実装が必要です。

// Java (CometD Client Library の概念的な使用例)
import org.cometd.client.BayeuxClient;
import org.cometd.client.transport.ClientTransport;
import org.cometd.client.transport.LongPollingTransport;
import org.eclipse.jetty.client.HttpClient;
import java.util.HashMap;
import java.util.Map;

public class SalesforceCDCSubscriber {

    public static void main(String[] args) throws Exception {
        // 1. Salesforce への認証情報を設定
        String accessToken = "YOUR_ACCESS_TOKEN"; // OAuth 2.0 フローなどで取得
        String instanceUrl = "https://YOUR_INSTANCE.salesforce.com";
        String apiVersion = "58.0"; // 最新の API バージョンを使用

        // 2. HTTP クライアントを初期化
        HttpClient httpClient = new HttpClient();
        httpClient.start();

        // 3. CometD クライアント (BayeuxClient) を初期化
        Map options = new HashMap<>();
        options.put(ClientTransport.JSON_CONTEXT_OPTION, new org.cometd.client.ext.JacksonJSONContextClient());
        ClientTransport transport = new LongPollingTransport(options, httpClient);

        BayeuxClient bayeuxClient = new BayeuxClient(instanceUrl + "/cometd/" + apiVersion, transport);

        // 4. 認証ヘッダーを設定
        bayeuxClient.setHeaders(new HashMap() {{
            put("Authorization", "Bearer " + accessToken);
        }});

        // 5. CometD への接続
        bayeuxClient.handshake();
        bayeuxClient.waitFor(1000 * 5, BayeuxClient.State.CONNECTED); // 接続を待機

        if (!bayeuxClient.isDisconnected()) {
            System.out.println("CometD サーバーに接続しました。");

            // 6. 購読するチャンネルを指定
            // /data/ChangeEvents はすべての変更イベントを購読するためのチャンネルです。
            // 特定のオブジェクトの変更イベントは /data/AccountChangeEvent のように指定します。
            String topic = "/data/AccountChangeEvent";

            // 7. チャンネルを購読し、イベントリスナーを登録
            bayeuxClient.getChannel(topic).subscribe((channel, message) -> {
                System.out.println("受信した変更イベント: " + message.getData());
                // ここで受信したイベントデータをパースし、ビジネスロジックを実行
                // message.getData() は JSON 形式のイベントペイロードです。
            });

            System.out.println("チャンネル " + topic + " を購読しました。イベントを待機中...");
            // プログラムを終了させないために無限ループまたは適切な終了ロジック
            Thread.sleep(Long.MAX_VALUE); // この例では無限に待機
        } else {
            System.err.println("CometD サーバーへの接続に失敗しました。");
        }

        // 8. クライアントをシャットダウン (通常はアプリケーション終了時)
        // bayeuxClient.disconnect();
        // httpClient.stop();
    }
}

実装ロジック解析:

認証情報の取得: 外部システムは OAuth 2.0 などの標準的な方法で Salesforce からアクセストークンを取得する必要があります。
CometD クライアントの初期化: Jetty の HTTP クライアントと CometD クライアントライブラリ (BayeuxClient) を使用して、Salesforce の Streaming API エンドポイントに接続します。
認証ヘッダーの設定: 取得したアクセストークンを HTTP ヘッダーとして設定し、Salesforce に対して認証を行います。
ハンドシェイクと接続: bayeuxClient.handshake() メソッドで CometD サーバーとの接続を確立します。
チャンネル購読: 変更イベントは特定のチャンネルを通じて配信されます。/data/AccountChangeEvent のように、/data/ の後に変更イベントオブジェクト名を指定して購読します。汎用的な変更イベントは /data/ChangeEvents で購読できますが、これは通常推奨されません。
イベント処理: 購読が成功すると、指定したリスナー関数が新しいイベントを受信するたびに呼び出されます。message.getData() からは JSON 形式のイベントペイロードを取得でき、これをパースしてビジネスロジックに利用します。

注意事項とベストプラクティス

権限要件

変更イベントを購読するユーザーまたは統合ユーザーには、「Subscribe to Change Events」権限セットライセンスとその権限が割り当てられている必要があります。
特定のオブジェクトの変更イベントのみを購読する場合、そのオブジェクトに対する「参照 (Read)」権限が必要です。

Governor Limits (2025年最新版)

変更イベントの発行制限：各 Salesforce 組織は、1日あたり最大 100,000 件の変更イベントを発行できます（Developer Edition は 50,000 件）。追加ライセンスを購入することで拡張可能です。
Streaming API (CometD) の同時接続数：1つの Salesforce 組織あたり最大 2,000 の同時 Streaming API 接続が可能です。
Platform Event Trigger の非同期実行制限：Platform Event Trigger は非同期 Apex として実行されるため、通常の非同期 Apex と同様に、1日あたり最大 250,000 回の非同期 Apex メソッド実行（または組織の購入した制限）にカウントされます。
イベントストリームの保持期間：イベントはデフォルトで72時間保持されます。この期間内であれば、過去のイベントを再生購読（Replay Subscription）することも可能です。

エラー処理

再試行メカニズム：外部システムがイベントを受信できなかった場合（ネットワーク障害など）、堅牢な CometD クライアントは自動的に再接続とイベントの再試行を試みるべきです。
デッドレターキュー (Dead-Letter Queue)：イベント処理に失敗した場合は、デッドレターキューにイベントを格納し、後で手動または自動で再処理できるようなメカニズムを実装することを推奨します。
順序保証：CDC は発行されたイベントの順序を保証しますが、外部システムの処理順序は保証されません。

パフォーマンス最適化

フィルタリング：Streaming API で購読する際に、SOQL WHERE 句に似た構文でイベントをフィルタリングすることができます。これにより、不要なイベントの受信を防ぎ、クライアント側の処理負荷を軽減できます。
バルク処理：Platform Event Trigger で Apex コードを記述する際は、トリガーのベストプラクティスに従い、単一イベントだけでなくイベントのリスト全体をバルクで処理するロジックを実装してください。
イベントペイロードの最適化：不要なフィールドは変更データキャプチャの対象から除外することで、イベントのサイズを小さくし、ネットワーク帯域と処理時間を節約できます。

よくある質問 FAQ

Q1：Change Data Capture (CDC) とカスタム Platform Event の主な違いは何ですか？

A1：CDC は、既存の Salesforce オブジェクトの DML 操作を自動的にイベントとして発行する機能です。開発者はイベントを発行するためのコードを記述する必要がありません。一方、カスタム Platform Event は、開発者が独自のビジネスイベントを定義し、Apex の EventBus.publish() メソッドを使って明示的に発行するものです。CDCは「何が変更されたか」を、カスタム Platform Event は「何が起こったか」を表現するのに適しています。

Q2：変更イベントのデバッグはどのように行いますか？

A2：Salesforce 内部で Platform Event Trigger を使用している場合は、通常の Apex トリガーと同様に「デバッグログ」でログレベルを設定して確認できます。外部の CometD クライアントを使用している場合は、クライアント側のログ出力と Salesforce の「デバッグログ」で Streaming API の接続状況やエラーメッセージを確認します。また、Salesforce CLI の sfdx event:listen コマンドを使用して、リアルタイムでイベントストリームを監視することも可能です。

Q3：CDC のパフォーマンスを監視するための主要な指標は何ですか？

A3：主要な指標としては、イベントの発行数と受信数（「設定」の「プラットフォームイベントの利用状況」で確認可能）、Streaming API の同時接続数、イベントの遅延（イベント発生から受信までの時間）、および Platform Event Trigger の Apex 実行時間とガバナ制限違反の有無が挙げられます。これらの指標を監視ツールで追跡し、パフォーマンスのボトルネックを特定することが重要です。

まとめと参考資料

Salesforce Change Data Capture は、今日の統合されたビジネス環境において、リアルタイムに近いデータ同期を実現するための不可欠なツールです。開発者として CDC を活用することで、Salesforce を中心としたエコシステム全体のデータ鮮度とビジネスプロセスの応答性を大幅に向上させることができます。疎結合なアーキテクチャ、効率的なリソース利用、そして堅牢なエラー処理を念頭に置いた設計が成功の鍵となります。

公式リソース

📖 公式ドキュメント：Change Data Capture Developer Guide
📖 公式ドキュメント：Streaming API Developer Guide
🎓 Trailhead モジュール：Platform Events Basics
🎓 Trailhead モジュール：Platform Events Advanced

にゃんこのSalesforce日記

このブログを検索