Confluent（Kafka REST）

Amplitude DataのConfluent連携は、Confluent Kafka REST v3プロデュースAPIを使用してAmplitudeイベントデータをKafkaトピックにストリーミングします。これを使用して、独自のコネクターを実行することなく、AmplitudeイベントをConfluent Cloudまたは自己管理型のConfluent RESTプロキシに送信できます。

考慮事項

• この送信先はConfluentのKafka REST v3 APIに対応しています。これは Confluent Cloud や、自己管理型の Confluent REST プロキシまたは Confluent サーバーと連携します。 これは古い v2 ルートのみを提供する Redpanda や、HTTP プロデュース API をまったく公開していない Amazon MSK やオープンソースの Apache Kafka では動作しません。
• REST エンドポイントは、パブリック インターネットに接続し、パブリック IP アドレスに解決する必要があります。 Amplitudeはlocalhost、内部専用のホスト名およびプライベートアドレス範囲を拒否するため、この送信先をVPC内部プロキシやローカルにホストされているRESTプロキシに向けることはできません。
• Amplitudeは、Amplitude Export APIがドキュメント化したものと同じJSONをレコードvalueに格納します。
• Confluent はほぼリアルタイムのストリーミング送信先であるため、イベントストリーミングの使用状況にカウントされます。
• この送信先は ID 転送をサポートしています。AmplitudeにIdentifyコールを発信するたびに、Amplitudeはそのユーザー情報を転送します。 詳細については、Identify のドキュメントを参照してください。
• Amplitudeはリクエストごとに1レコードを生成します。 Confluent は非ストリーミングプロデュースモードを、接続あたり 1 秒あたり 100 リクエスト未満で記録しているため、この送信先にルーティングするボリュームをそれに応じて計画してください。
• AmplitudeはAPIキーとシークレットを静止状態で暗号化し、それらをすべてのプロデュースリクエストでHTTP基本認証資格情報として送信します。

連携を設定する

Confluent の設定

Amplitudeで送信先を作成する前に、Confluent Cloud（または自己管理型クラスター）でこれらの手順を完了してください。

1. トピックを作成する

Amplitudeイベントを受信するトピックを作成するか、既存のトピックを選択します。Amplitudeはトピックを作成しません。そのため、クラスターでトピックの自動作成が有効になっている場合を除き、Amplitudeが最初のイベントを送信する前にトピックが存在している必要があります。

予想されるボリュームに基づいてパーティション数を選択してください。Amplitudeはすべてのイベントにレコードキーを設定するため、同じユーザーのイベントは常に同じパーティションに送信されます。

2. サービスアカウントとAPIキーを作成する

1. Amplitudeのサービスアカウントを作成します。
2. そのサービスアカウント用の Kafka API キーとシークレットを作成します。このキーとシークレットは、トピックをホストするクラスターにスコープされます。
3. シークレットを安全な場所に保管してください。Confluent ではこの値を 1 回しか表示しません。

3. トピックへの書き込みアクセスを許可する

サービスアカウントに、トピックへのプロデュース権限を付与します。Confluent CLI を使用する場合:

confluent kafka acl create \
  --allow \
  --service-account sa-xxxxxx \
  --operations write \
  --topic your-topic-name \
  --cluster lkc-xxxxxx \
  --environment env-xxxxxx

Amplitudeはプロデュースのみを行うため、送信先が必要とする操作はこれだけですwrite。

4. 接続の詳細を収集する

Amplitudeに入力する次の値を収集します。

エンドポイントに API バージョンを含めます。 Confluent Cloud と Confluent Server は /kafka/v3を提供し、スタンドアロン REST プロキシは /v3を提供しています。Amplitude はクラスタとトピックパスのセグメントのみを追加します。

Amplitude設定

1. Amplitudeデータで、**[カタログ]をクリックし、[宛先]**タブを選択します。
2. 「イベントストリーミング」セクションで、「Confluent (Kafka REST)」をクリックします。
3. 同期名を入力し、[同期を作成] をクリックします。
4. ［Edit］ をクリックし、次のように入力します。
   • Kafka REST エンドポイント: Confluent からのエンドポイント。サフィックス/v3または /kafka/v3を含む。
   • Kafka クラスター ID: 例: lkc-xxxxxx。
   • トピック：イベントを受信するトピックです。
   • API キーと API シークレット: サービスアカウント用に作成した認証情報です。 REST プロキシが認証なしで実行されている場合に限り、両方を空白のままにしてください。
   • レコードキーとして使用するイベントフィールド：オプションです。 デフォルトを使用するには空白のままにしておきますuser_id。
5. ［イベントを送信］ フィルタを使用して、送信するイベントを選択します。 すべてのイベントを送信できますが、Amplitudeでは最も重要なイベントを選択することをお勧めしています。
6. 完了したら、送信先を有効にして保存します。

記録形式

Amplitude はイベントごとに 1 つのプロデュースリクエストを送信先に送信します。ボディは以下の{endpoint}/clusters/{clusterId}/topics/{topic}/recordsようになります：

{
  "key": {"type": "STRING", "data": "user-123"},
  "value": {"type": "JSON", "data": {"user_id": "user-123", "event_type": "Checkout Completed", "...": "..."}}
}

valueこれは、Export API スキーマと一致する、JSON オブジェクトとしての完全な Amplitude イベントです。

これはkeyパーティショニングを制御します。Amplitudeは、イベントフィールドで指定されたフィールドをレコードキーとして使用します。その設定が空白の場合は、デフォルトuser_idを使用します。イベントで選択されたフィールドが欠落しているか空の場合、Amplitude は device_id にフォールバックし、匿名トラフィックを安定したパーティションに維持します。 どちらのフィールドにも値がない場合、Amplitude はそのキーを省略し、Confluent はそのパーティションを割り当てます。

トラブルシューティング