TL;DR

Salesforce Data CloudのCRMコネクタは、データバンドル利用時にはバッチ取り込みが一般的ですが、実はストリーミング取り込みも可能です。これにより、CRMで発生したデータ変更をほぼリアルタイムでData Cloudに反映できます。本記事では、バッチとストリーミングの違い、フル更新と差分更新の概念を整理し、実際にストリーミング取り込みを設定・検証した手順を画面ショット付きで解説します。リアルタイムでの顧客プロファイル更新や迅速なセグメント更新など、Data Cloudの活用シーンを広げたい方は必見です！

CRMコネクタの取り込みモード整理

まず、CRMコネクタには以下の2つの取り込みモードがあります。

• バッチモード (Batch Mode)
  一般的にData Bundleを利用した場合のモード。
  10分間隔でCRMからデータを抽出・更新します。効率的ですがリアルタイムではありません。

• ストリーミングモード (Streaming Mode)
  CRMのレコード作成や更新をほぼリアルタイムにData Cloudに反映します。
  リアルタイム性が求められるケースで活用されます。

では、データストリームの設定時にどこでそれを設定するのか？
実はユーザーは明示的にバッチモードとストリーミングモードを選べるわけではありません。ストリーミングになるには以下の条件があり、つまり、該当する場合はバッチモードになるということです。

ストリーミングが使えないケース

1. オブジェクトや項目がストリーミング非対応の場合
CRMのすべてのオブジェクトがストリーミング取り込みに対応しているわけではありません。非対応のオブジェクトや、オブジェクトタイプ、派生項目（Derived Fields）を指定した場合、ストリーミングは利用できず、自動的にバッチモードになります。
参照: CRM データストリームのオブジェクトサポート

2. 数式項目（Formula Field）を選択した場合
CRMオブジェクトに設定された数式項目を、Data Cloudのデータストリーム項目として選択すると、ストリーミング取り込みは利用できず、自動的にバッチモードに切り替わります。これは数式項目がリアルタイム更新イベントをトリガーできないためです。

3. Salesforce CRM コネクタの「絞り込み機能」を使っている場合
データストリームに検索条件を定義している場合、そのデータストリームはストリーミングモードを利用できません。絞り込み条件によるデータ取得は、仕組み上バッチ（一括処理）での取り込みになります。

4. 「Data Cloud Salesforce コネクタ」権限セットを変更した場合
Data CloudとCRM組織を接続すると、自動的に「Data Cloud Salesforce コネクタ」という権限セットが作成されます。これはCRMからData Cloudへのデータ転送に使用する「プラットフォームインテグレーションユーザー（Integration User）」向けのものです。
この権限セットに通常のCRMユーザーを追加するなど、設定を変更するとストリーミングに必要な権限が正しく機能しなくなり、結果的にストリーミングモードが使えずバッチモードでの取り込みになってしまいます。

リアルタイム連携をしたいなら、必ずこれらの条件に注意してデータストリーム設計を行いましょう。

参照: CRM コネクタストリーミング

フル更新、差分更新とストリーミングの関係

CRMコネクタにはバッチモードの時には「フル更新」と「差分更新」の概念もあります。

- フル更新（Full Refresh）
全データを定期的に再取り込みします。データの棚卸し的なイメージです。

- 差分更新（Incremental Refresh）
前回の取り込み以降に変更されたデータのみ取り込みます。

参照:
抽出のデータ更新プロセス
データストリームの設定と更新モード

ストリーミングモードは、この差分更新をリアルタイムで実施する仕組みです。CRM側の変更を即時に反映できるため、最新のデータを常に維持できます。バッチモードにおける差分更新とは、リアルタイム性という点で大きく異なります。

ストリーミングモードの裏側では何が動いているのか？

ストリーミングモードでリアルタイムにデータが取り込まれる仕組みの背景には、SalesforceのChange Data Capture（CDC）という機能があります。

CDCは、Salesforce内でレコードが作成・更新・削除された際に、それをイベントとして通知してくれる機能です。CRMコネクタのストリーミングモードは、このCDCの仕組みを使って、CRM側の変更をリアルタイムでData Cloudに伝えています。

参照: CDCの概要

つまり、ストリーミングモードとは「差分更新をリアルタイムで実現するCDCベースの取り込み」だと捉えると、理解しやすくなります。

なお、CDCが有効になっていないオブジェクトや環境では、ストリーミングは動作しないため、Data Cloud側では自動的にバッチモードに切り替わることがあります。

ストリーミング取り込みを試してみる

実際にデータストリームを設定し、対象となるCRMオブジェクトを作成したり変更してみて、どれくらいのタイミングで変更が反映されるかをみてみます。

CRMコネクタを使った新規データストリームの作成

まずデータストリームを作成していきます。
データストリームタブに遷移し、画面右上にある「新規」ボタンをクリックして新規のデータストリームを作成していきます。

Salesforce CRMを接続ソースとして選び、「次へ」をクリックします。

新規データストリームの画面ではデータバンドルが最初選ばれていますが、「すべてのオブジェクト」をクリックしてから、取り込みたいオブジェクトを選びます。
今回はわかりやすく取引先責任者を対象のオブジェクトとします。

Standard FieldsとCustom Fieldsのタブをクリックして、取り込みのフィールドを選びます。
今回は以下の項目を選んでみました。

作成したデータストリームを開いてみます。
詳細タブをクリックし、頻度の箇所を確認します。
「ストリーミング」になっていればストリーミングモードでデータが取り込めます。

では、データを取り込みを実施していきます。
作成したデータストリームを選択して、状況の更新をクリックして実行します。

最終事項状況が成功になったらデータが取り込まれています。
更新履歴タブをクリックして、履歴を確認します。
初回は完全置換モードでフルロードされます。

クエリエディタでデータの中身を確認してみます。利用するクエリは取り込んだ値や、変更した値を確認できるようにフィールドを選んでください。

SELECT "Id__c",
	   "LastName__c",
	   "FirstName__c",
	   "Email__c",
	   "Phone__c",
	   "LastModifiedDate__c",
	   "SystemModstamp__c" 
FROM "Contact_Home__dll" 
ORDER BY "SystemModstamp__c" DESC

以下のようにデータレイクオブジェクトに値が取り込まれていることが確認できます。

では実際に、CRM側のデータを変更してみます。
Oscar Tranさんの電話番号を変えてみたいと思います。

元は(415) 555-1212となっているので、(415) 555-1111に変更して保存します。

保存後すぐにデータが変更されているか確認するため、クエリエディタで先ほどのクエリを実行します。

すると、Oscar Tranさんの電話番号がすぐに変更されていることが確認できます。
※もし変更されない場合はちょっと待ってみてください。

最後に

今回の検証を通じて、CRMコネクタのストリーミングモードが、Salesforce CRMでのデータ変更をほぼリアルタイムでData Cloudに取り込めることを確認しました。Change Data Capture（CDC）を活用したこの仕組みは、顧客プロファイルや行動データを即時に活用したいユースケースにおいて、非常に有効です。

ただし、ストリーミング取り込みは常に使用できるわけではなく、対象オブジェクトや項目の種類、設定状況によっては自動的にバッチモードへ切り替わる点に注意が必要です。また、Data Cloudでは1つのオブジェクトにつき1つのデータストリームしか定義できないため、バッチとストリーミングの併用は構成上できません。

このため、設計段階では「どのデータにリアルタイム性が本当に必要か」を慎重に見極めたうえで、適切にストリーミングとバッチを使い分けることが求められます。

とはいえ、バッチモードでも10分程度の取り込み間隔があり、多くの業務ユースケースでは十分な更新頻度です。リアルタイム性が強く求められるケースにはストリーミングを活用し、それ以外は安定性や汎用性の高いバッチモードで対応するのが、現時点のData Cloudにおける現実的かつバランスの取れたアプローチといえるでしょう。

参照: Salesforce CRM コネクタ

※本記事は、私が所属する会社とは一切関係のない事柄です。