Stream Platform Events directly from Salesforce into Data Cloud using the Ingestion API

Salesforce プラットフォームのイベントをデータクラウド取り込み API に直接ストリーミングする完全な「オンプラットフォーム」アプローチ。

Data Cloud にリアルタイムでストリーミングしたい複数のプラットフォームイベントタイプがあるとします。これは複数の組織からのものであり、複数のデータクラウドインスタンスにストリーミングされる可能性があります。

これには、柔軟で再利用可能なソリューションが必要です。これは、あらゆる種類のプラットフォームイベントに使用して、任意のデータクラウドインジェストストリーミング API にマッピングできる単一の実装です。そこで、ミニイベントブリッジを作ることにしました…

さらに… プラットフォームイベントオブジェクトをマッピングできたら、あらゆる種類の Salesforce オブジェクトにまったく同じロジックを実装できます。これは、フロー、LWC、Apex が最小限のコードでイベントをデータクラウドに直接ストリーミングできるため、非常に優れています。

要件

正確に達成したいことの概要を説明することから始めましょう。

任意のタイプのプラットフォームイベントまたは sObject を任意の Data Lake オブジェクトにマップします
複数の組織から再利用できる構成可能なセットアップ
必要なコードは最小限でありながら非常に柔軟なソリューション

プロセスは次のようになります。

私たちは何が必要なのか？

セキュリティ上の理由から名前付き認証情報を利用する、任意のスポーク組織から 1 つ以上のデータクラウドインスタンスへの安全な接続。これを実現するためにカスタム認証プロバイダーパッケージを
作成しました。設定方法については説明書をご覧ください。

sObject フィールドと取り込み API オブジェクトフィールド間のマッピングを保持するカスタムメタデータ設定。
構成では、ターゲットの取り込み API コネクタとターゲットのオブジェクト名も追跡する必要があります。
最後に、名前付き資格情報を定義する必要があります。
Ingestion API を使用すると、複数のレコードを同時に取り込むことができるため、Apex のベストプラクティスに従って、ソリューションを一括化する必要があります。
コードは次のようになります。

trigger PlatformEventTrigger on  PlatformEvent ( after insert ) { 
    
  // ユーザー定義の構成に基づいてイベント レコードをストリーミングします
  // このロジックは任意のクラス/メソッドから呼び出すことができます
  // Async メソッドはトリガーのみに必要です
  // これがすべてのコードですエンド ユーザーは
  utl.Dc.streamRecordsToDataCloudAsync( '[CONFIG_NAME]' ,trigger.New);について心配する必要があります。

考慮すべき限界

インジェスト API ストリーミングペイロードは 200 kb を超えることができないため、大量のレコードを送信することはできません。ただし、通常は 1 つだけである必要があります。
Ingestion API を呼び出すと、データクラウドの制限が侵食されます。
アウトバウンドコールは Salesforce の API 制限を消費しません
apex トリガーを使用してプラットフォームイベントをサブスクライブしても、サブスクライバーの制限にはカウントされません
プラットフォームイベントの作成は、引き続き (時間ごとの) 制限にカウントされます
Apex トリガーを使用するため、Data Cloud への API 呼び出しは非同期で起動する必要があります。これは将来のメソッドを使用します。将来のメソッドがいつ起動されるかは、リソースの可用性に基づいています。通常、これらは即座に実行されますが、理論的には数分かかる場合があり、これはほぼリアルタイムのソリューションとなります。ストリーミング取り込み API は 2 分ごとにレコードを処理するため、これが実際の問題であるとは考えませんが、留意してください。

プラットフォームのイベントトリガーコールアウトアーキテクチャに関する注意事項

Web サービス呼び出しを介してイベントを中継するのは、少しアンチパターンであると考えられます。Cometd や gRPC のような (比較的複雑な) サブスクライバクライアントが必要なく、サブスクライバの制限を侵食しないという利点がありますが、いくつかの欠点もあります。
コールアウトが失敗した場合、ReplayId を利用して送信されたイベントを再度取得する方法はありません。ソリューションの堅牢性がどの程度必要かによっては、カスタムのエラー処理および再試行メカニズムの実装が必要になる場合があります。私は個人的に、多くのシナリオではメリットがデメリットを上回ると考えていますが、この種の実装の影響については十分に認識してください。

重要

セキュリティは簡単なテーマではありません。この (または他の) ソリューションを実装する前に、認定されたセキュリティの専門家および認定された実装パートナーと協力して、常に何を行っているかを検証してください。

TL;DR :: データクラウドユーティリティ

マッピングオブジェクトと、Apex または Apex トリガーからデータクラウドにレコードをストリーミングするための同期および非同期メソッドを含むデータクラウドユーティリティを作成しました。
このユーティリティには、一括取り込みと YAML 作成用のツールセットも付属しています。詳細なセットアップ全体を通じて、ユーティリティのメソッドを参照します。

G itHubリポジトリはここにあります:
https://github.com/jfwberg/lightweight-data-cloud-util

ステップバイステップのセットアップ例

さて、話は十分にしましたので、例を設定しましょう。この例では、スマートデバイスイベントを表すプラットフォームイベントがあります。これには、照明を点灯したり、電力が安い時間に洗濯機を稼働させるようにスケジュールを設定したり、温度が設定されたしきい値を下回ったときに暖房を有効にしたりすることができます。

注: この例全体はプラットフォームイベントに基づいていますが、あらゆる種類の Salesforce オブジェクトや、フローや LWC で簡単に動作させることができる Apex でも動作します。

完全に構成されたデモは、Git リポジトリで入手できます。

01 :: スポーク組織 :: パッケージをインストールする

更新: セットアップに関する詳細なステップバイステップガイドをここに書きました: Salesforce Data Cloud Utility および Ingestion Api UI のセットアップ手順これらの手順の最初のステップにリストされているすべてのパッケージをインストールしてください。

02 :: 接続アプリと名前付き認証情報をセットアップする

接続するには、スポーク組織に名前付き認証情報を設定し、データクラウド組織に接続アプリを設定する必要があります。指示に従ってください。
複数のスポーク組織または複数のデータクラウドインスタンスが必要な場合は、すべてのターゲットで手順を繰り返すだけです。

03 :: スポーク組織 :: プラットフォームイベントを作成する

セットアップでは、プラットフォームイベントを作成します。この例では、イベントは Smart_Device_Event__e と呼ばれ、スマートホームデバイスから送信されるイベントを表します。
実例を示すためにいくつかのフィールドを作成します。Action__c はオンになっていますか、それともオフになっていますか? Device_Type__c (つまり電球)、数値フィールドをテストするためのStatus_Code__c 、および日付/時刻フィールド値をテストするためのTimestamp__c 。
イベントを一意にするために、EventUuIdフィールドを使用できます。プッシュ動作を「すぐに公開」に設定します。

04 :: データクラウド :: インジェスト API コネクタの作成

データクラウドを開き、設定歯車からデータクラウドのセットアップに移動します
左側のメニューバーで「Ingestion API」ボタンをクリックします
右上隅にある「新規」ボタンをクリックし、取り込みコネクタに名前を付けます。取り込みソースはプラットフォームイベントからのスマートデバイスイベントになるため、これを「スマートデバイスイベントコネクタ」と呼びます。

こで、イベントデータをマップするオブジェクトを表す YAML ファイルを作成します。フォーマットの詳細については、ドキュメントを参照してください。ファイルをドライブ上のどこかに保存します。ユーティリティには SObject 2 YAML ジェネレーターがあることに注意してください。それも活用できます。

openapi: 3.0.3
components:
  schemas:
    Smart_Device_Event:
      type: object
      properties:
        ReplayId:
          type: string
        EventUuid:
          type: string
        Action:
          type: string
        Device_Type:
          type: string
        Status_Code:
          type: number
        Status_Reason:
          type: string
        Timestamp:
          type: string
          format: date-time

セットアップでは、新しく作成した取り込み API コネクタの詳細ページで [ファイルをアップロード] ボタンを押して、YAML ファイルをアップロードします。フィールドのプレビューが表示されるはずです。「保存」を押します
。これらのフィールド名は、後でマッピング設定で設定する予定のフィールド名と同じなのでメモしてください。

05 :: データクラウド :: 新しいデータストリームを作成する

これで初期設定は完了です。Data Cloud アプリに戻り、[データストリーム] タブに移動します。
「新規」をクリックし、「取り込みAPI」タイルを選択して「次へ」をクリックします。
作成したばかりの取り込み API を選択し、YAML ファイルからオブジェクトを選択します。「次へ」をクリックしてください

希望するイベントカテゴリの種類を選択します。この場合はエンゲージメントデータなので、[エンゲージメント]を選択します。プライマリにはイベントの一意の識別子であるEventUuidフィールドを選択し、タイムスタンプにはカスタムのTimestampフィールドを選択します。準備ができたら「次へ」をクリックします

必要に応じてデータスペース名を設定するか、「デフォルト」のままにして「デプロイ」を押します。
これで、データクラウドがすべてセットアップされ、ストリーミングデータを受信する準備が整いました。

06 :: スポーク組織 :: 構成とマッピングを作成する

次に、スポーク組織に戻り、カスタムメタデータを設定して、設定を作成し、カスタムメタデータレコードをマッピングします。

「Data Cloud Ingestion API Configuration」カスタムメタデータタイプの横にある「Manage Records」をクリックし、「New」をクリックします。
ラベルとして、プラットフォームイベントに一致するものを選択します。この場合は「スマートデバイスイベント構成」です。この API 名が、後で Apex コードで識別子として必要になるわけではありません。
[名前付き認証情報名] で、データクラウドの名前付き認証情報の API 名を指定します。(動作テストを行ったことを確認してください)。複数のデータクラウドインスタンスにストリーミングする場合は、これらすべてのレコードを複製する必要があります。これは一般的な使用例ではないと思うので、特にそのために何も構築していません。
インジェスト API コネクタ名を、先ほど設定したコネクタ名に設定します。この場合は「Smart_Device_Event_Connector」
最後に、ターゲットオブジェクト名を、取り込み API / YAML ファイルで指定されている名前に設定します。この場合は「Smart_Device_Event」です。

「保存」を押します
[設定] のカスタムメタデータタイプに戻り、「データクラウドインジェスト API フィールドマッピング」メタデータタイプの横にあるレコードの管理をクリックします。
「新規」をクリックします
フィールドごとにマッピングレコードを作成します。マッピング名が一意のキーであることを確認してください (残念ながら、これは必須であり、手動です)
ソース (プラットフォームイベントフィールド) とターゲット (取り込み API コネクタフィールド) を選択し、それに応じて構成レコードへのルックアップを設定します。

すべて完了すると、マッピングテーブルは次のようになります。

07 :: スポーク組織 :: トリガーの作成

設定が完了したので、プラットフォームイベントにトリガーを作成します。

trigger SmartDeviceEventTrigger on Smart_Device_Event__e  (after insert) { 
    
    // データをデータ クラウドにストリーミングします
    utl.Dc.streamRecordsToDataCloudAsync( 
        JSON.serialize(trigger.new), 
        'Smart_Device_Event_Configuration'
     ); 
}

トリガーではすべて、utl.Dc.streamRecordsToDataCloudAsync()メソッドを呼び出します。これは将来的な方法です。(したがって非同期です)
最初の引数は、プラットフォームイベントの JSON シリアル化バージョンです。これは、future メソッドがオブジェクトを引数として受け取ることができないためです。ということで往復連載をします。非同期メソッドを使用する場合は、sObject リストを渡します。
2 番目の引数はメタデータ構成レコードの名前であり、おそらくカスタムラベル内にある必要があります。しかし、デモの場合はこれで十分です。

08 :: スポーク組織 :: PlatformEventSubscriberConfig メタデータを作成する

外部資格情報では、名前付き資格情報を介して外部資格情報を呼び出すユーザーが、プロファイルまたは権限セットを通じて名前付きプリンシパルにアクセスできる必要があります。
残念ながら、プラットフォームイベントは、アクセス権を割り当てることができない自動化されたユーザーとして実行されるため、将来のメソッドを通じてコールアウトを行うことは不可能になります。

この問題を解決する唯一の方法は、プラットフォームイベントを実行する実行 (システム) ユーザーを指定することです。これは、PlatformEventSubscriberConfigメタデータタイプを使用して行われ、セットアップを通じて行うことはできません。メタデータ API を通じて行う必要があります。
リポジトリ内のデモフォルダーにはサンプルが含まれています。デプロイする前に必ずユーザー名を更新してください。

実行中のユーザー情報と手順 07 で作成したプラットフォームイベントトリガーの名前を含む XML ファイルを作成します。
ファイルにSmartDeviceEventTrigger.platformEventSubscriberConfig-meta.xml という名前を付け、メタデータ API または VSCode を使用してデプロイします。

<?xml version= "1.0" encoding= "UTF-8" ?> 
< PlatformEventSubscriberConfig  xmlns = "http://soap.sforce.com/2006/04/metadata" > 
    < platformEventConsumer > SmartDeviceEventTrigger </ platformEventConsumer > 
    <バッチサイズ> 100 </batchSize> <masterLabel> SmartDeviceEventTriggerConfig </masterLabel> <user> 
    test-382rc1gvovzk@example.com </user> <isProtected> false
     </isProtected> </PlatformEventSubscriberConfig>

プラットフォームイベントで [サブスクリプション]に移動し、[管理] をクリックして [一時停止]をクリックし、新しいユーザーを有効にすることが重要です。デモのようにすべてを一度にデプロイすると、すぐに機能するはずです。しかし、それが機能しない理由を理解することで、頭痛の種を軽減できるかもしれません。実行中のユーザーが外部認証情報にアクセスできることを確認してください。

09 :: スポーク組織 :: テスト

これですべての設定が完了しました。あとは、いくつかのプラットフォームイベントを起動して、それらがデータクラウドにストリーミングされるのを監視するだけです。イベントを公開する例を次に示します。EventUuid と ReplayId は自動的に設定されます。

EventBus.publish(
    new Smart_Device_Event__e(
        Action__c        = 'Switch State',
        Device_Type__c   = 'Lightbulb',
        Status_Code__c   = 1,
        Status_Reason__c = 'Switched On (1)',
        Timestamp__c     = Datetime.now()
    )
);

データクラウドで[データエクスプローラー]タブに移動し、選択したデータスペースでスマートデバイスイベントコネクタデータレイクオブジェクトを選択すると、イベントがストリーミングされているのを確認します。

10 :: スポーク組織 :: 例

完全な構成例は GitHub リポジトリで入手できます。これには、プラットフォームイベントの発生をテストするための LWC が含まれています。試してみて、何ができるかをよりよく理解することができます。

結論

これで完了です。これは、カスタムメタデータ構成を使用して、1 行のコード (およびいくつかのパッケージマジック) でデータを Data Cloud にストリーミングする方法の明確な例です。

セットアップには少し時間がかかりますが、柔軟性があり、定型的なコードを記述する必要がなく、接続は名前付き資格情報によって処理されます。

Ingestion API を使用して、プラットフォーム イベントを Salesforce から Data Cloud に直接ストリーミングします

要件