24.4 Kafka ConnectとAdobe Experience Platform Sink Connectorのインストールと設定

24.4.1 Adobe Experience Platform Sink Connectorのダウンロード

https://github.com/adobe/experience-platform-streaming-connect/releasesに移動し、Adobe Experience Platform Sinkコネクタの最新の正式リリースをダウンロードします。

カフカ

ダウンロードファイル​streaming-connect-sink-0.0.6-java-11.jar​をデスクトップに配置します。

カフカ

24.4.2 Kafka Connectの設定

デスクトップ上の​Kafka_AEP​という名前のフォルダーに移動し、kafka_2.13-2.8.0/configフォルダーに移動します。
そのフォルダーで、任意のテキストエディターを使用して​connect-distributed.properties​ファイルを開きます。

カフカ

テキストエディターで、34行と35行に移動し、key.converter.schemas.enableとvalue.converter.schemas.enableをfalseに設定します。

``json
key.converter.schemas.enable=false
value.converter.schemas.enable=false
``

カフカ

次に、フォルダーkafka_2.13-2.8.0に戻り、新しいフォルダーを手動で作成し、connectorsという名前を付けます。

カフカ

フォルダーを右クリックし、「Folder 」で「 New terminal」をクリックします。

カフカ

そうすれば、これが見える。 コマンドpwdを入力して、そのフォルダーの完全なパスを取得します。 フルパスを選択し、クリップボードにコピーします。

カフカ

テキストエディタに戻り、connect-distributed.properties​ファイルに戻り、最後の行まで下にスクロールします（スクリーンショットの86行目）。 # plugin.path=で始まる行のコメントを解除し、connectorsという名前のフォルダーへの完全なパスを貼り付ける必要があります。 結果は次のようになります。

plugin.path=/Users/woutervangeluwe/Desktop/Kafka_AEP/kafka_2.13-2.8.0/connectors

変更を​connect-distributed.properties​ファイルに保存し、テキストエディタを閉じます。

カフカ

次に、connectorsという名前のフォルダーにダウンロードしたAdobe Experience Platform Sink Connectorの最新の正式リリースをコピーします。 前にダウンロードしたファイルの名前は​streaming-connect-sink-0.0.6-java-11.jar​です。connectorsフォルダーに移動するだけで済みます。

カフカ

次に、kafka_2.13-2.8.0​フォルダのレベルで新しいターミナルウィンドウを開きます。 そのフォルダを右クリックし、[フォルダ​]の[新しいターミナル]をクリックします。

ターミナルウィンドウで、次のコマンドを貼り付けます。bin/connect-distributed.sh config/connect-distributed.propertiesをクリックし、​​と入力します。 このコマンドは、Kafka Connectを起動し、Adobe Experience Platform Sink Connectorのライブラリを読み込みます。

カフカ

数秒後、次のような画面が表示されます。

カフカ

24.4.3 Postmanを使用したAdobe Experience Platform Sinkコネクタの作成

Postmanを使用してKafka Connectを操作できるようになりました。 これを行うには、このポストマンコレクションをダウンロードし、デスクトップ上のローカルコンピューターに解凍します。 Kafka_AEP.postman_collection.jsonという名前のファイルが作成されます。

カフカ

このファイルはPostmanにインポートする必要があります。 それには、Postmanを開き、「読み込み」をクリックし、ファイルKafka_AEP.postman_collection.jsonをポップアップにドラッグ&ドロップして、「読み込み」をクリックします。

カフカ

このコレクションは、Postmanの左側のメニューに表示されます。 最初のリクエスト​GET利用可能なKafka Connectコネクタ​をクリックして開きます。

カフカ

そうすれば、これが見える。 青い「送信」ボタンをクリックします。このボタンをクリックすると、空の応答[]が表示されます。 空の応答は、Kafka Connectコネクタが現在定義されていないためです。

カフカ

コネクタを作成するには、Kafkaコレクションの2番目のリクエスト​POSTAEPシンクコネクタを作成​をクリックして開きます。 そうすれば、これが見える。 11行目に、"aep.endpoint"と表示されます。""と入力する場合は、演習24.3の最後に受け取ったHTTP APIストリーミングエンドポイントURLを貼り付ける必要があります。 HTTP APIストリーミングエンドポイントURLは次のようになります。https://dcs.adobedc.net/collection/94981e0634e0d37c3559ce7ece05a35eae35c52cc5962d2d4a44e488400f2338.

カフカ

貼り付けた後、リクエストの本文は次のようになります。 青い「送信」ボタンをクリックして、コネクタを作成します。 コネクタの作成がすぐに返されます。

カフカ

最初のリクエスト​GET利用可能なKafka Connectコネクタ​をクリックして再度開き、青い​送信​ボタンをもう一度クリックします。 これで、Kafka Connectコネクタが作成されました。

カフカ

次に、Kafkaコレクションの3番目のリクエスト​GETチェックKafka Connect Connect Status​を開きます。 青い​「​を送信」ボタンをクリックすると、次のような応答が返され、コネクタが実行中であることが示されます。

カフカ

24.4.4エクスペリエンスイベントの生成

kafka_2.13-2.8.0​フォルダーを右クリックし、「新しいターミナル」をクリックして、新しい​ターミナル​ウィンドウを開きます。

カフカ

次のコマンドを入力します。

bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic aep

カフカ

そうすれば、これが見える。 新しい行にEnterボタンを押すと、トピック​aep​に新しいメッセージが送信されます。

カフカ

メッセージを送信して、Adobe Experience Platformシンクコネクタで消費され、Adobe Experience Platformにリアルタイムに取り込まれるようになります。

これをテストするために、少しデモを行いましょう。

新しい匿名ブラウザーウィンドウを開き、https://public.aepdemo.netに移動します。

そうすれば、これが見える。

Launchの設定

設定IDを入力し、「設定を読み込む」をクリックします。 設定が読み込まれます。

Launchの設定

下にスクロールし、「設定を保存」をクリックします。

Launchの設定

その後、管理ホームページにリダイレクトされます。 「LDAP」を選択します。 LDAPを選択し、「保存」をクリックします。

Launchの設定

その後、管理ホームページにリダイレクトされます。 「ブランドを選択」に移動し、ブランド「Luma」を選択して、「保存」をクリックします。

Launchの設定

その後、管理ホームページにリダイレクトされます。 Luma​ロゴをクリックします。

Launchの設定

Lumaのホームページが表示されます。

Launchの設定

ログイン/登録​に移動します。 フォームに入力し、「アカウントを作成」をクリックします。 必ず​テストプロファイル​のチェックボックスをオンにしてください。

カフカ

X線パネルにプロフィールデータが表示されます。

カフカ

また、まだエクスペリエンスイベントが記録されていないこともわかります。

カフカ

それを変更し、KafkaからAdobe Experience PlatformにCallcenterのエクスペリエンスイベントを送信しましょう。

以下のエクスペリエンスイベントペイロードのサンプルを利用して、テキストエディターにコピーします。

{
  "header": {
    "datasetId": "5fd1a9dea30603194baeea43",
    "imsOrgId": "907075E95BF479EC0A495C73@AdobeOrg",
    "source": {
      "name": "Launch"
    },
    "schemaRef": {
      "id": "https://ns.adobe.com/experienceplatform/schemas/b0190276c6e1e1e99cf56c99f4c07a6e517bf02091dcec90",
      "contentType": "application/vnd.adobe.xed-full+json;version=1"
    }
  },
  "body": {
    "xdmMeta": {
      "schemaRef": {
        "id": "https://ns.adobe.com/experienceplatform/schemas/b0190276c6e1e1e99cf56c99f4c07a6e517bf02091dcec90",
        "contentType": "application/vnd.adobe.xed-full+json;version=1"
      }
    },
    "xdmEntity": {
      "eventType": "callCenterInteractionKafka",
      "_id": "",
      "timestamp": "2021-08-12T15:04:03.630Z",
      "_experienceplatform": {
        "identification": {
          "core": {
            "phoneNumber": ""
          }
        },
        "interactionDetails": {
          "core": {
            "callCenterAgent": {
              "callID": "Support Contact - 3767767",
              "callTopic": "contract",
              "callFeeling": "negative"
            }
          }
        }
      }
    }
  }
}

そうすれば、これが見える。 次の2つのフィールドを手動で更新する必要があります。

  • _id:例えば、ldapに設定し、その後にランダムな数字のシーケンスを指定します。vangeluw1234
  • phoneNumber:デモWebサイトで作成したアカウントのphoneNumberを入力します。X線パネルの​ID​の下にあります。
メモ

フィールド​**_id**​は、データの取り込みごとに一意である必要があります。 複数のイベントを生成する場合は、新しい一意の値に対して毎回フィールド​**_id**​を更新する必要があります。

カフカ

次のようになります。

カフカ

次に、完全なエクスペリエンスイベントをクリップボードにコピーします。 JSONペイロードの空白を削除する必要があります。そのためにオンラインツールを使用します。 http://jsonviewer.stack.hu/に移動してください。

カフカ

エディターにエクスペリエンスイベントを貼り付け、「空白を削除」をクリックします。

カフカ

次に、すべての出力テキストを選択し、クリップボードにコピーします。

カフカ

ターミナルウィンドウに戻ります。

カフカ

ターミナルウィンドウに、空白を含まない新しいペイロードを貼り付け、Enter​をクリックします。

カフカ

次に、デモWebサイトに戻り、ページを更新します。 これで、次のように、その他のイベント​の下に、プロファイルのエクスペリエンスイベントが表示されます。

カフカ

この練習は終わりました。

次の手順：概要と利点

モジュール24に戻る

すべてのモジュールに戻る

このページ