Apache Kafka コネクタ構成ガイド

構成

コネクタマーケットプレイスに移動し、新しいコネクタを追加します。コネクタを追加する一般的な手順については、コネクタについてを参照してください。

コネクタを追加した後、以下の構成を構成します：

ブートストラップサーバー
- （必須）エンドポイント用の Kafka ブートストラップサーバー（例：Conduktor Gateway ホストまたは直接 Kafka ブローカー：gateway.customer-domain.com:6969）。これは、PrivateLink、VPN、またはピアリングを介して Tealium からアクセス可能なエンドポイントを指す必要があります。
CA証明書（PEM）
- Kafka ブローカーまたはゲートウェイが提示する TLS 証明書を検証するための信頼アンカー。PEMエンコードされた CA チェーン。内部 PKI、プライベート CA、または自己署名証明書のみ必要です。
ホスト名検証の無効化
- Kafka クライアントのホスト名検証を切り替えます。SNI や非標準のホスト名を使用する場合は、この構成を無効にします。デフォルト：検証有効。
Confluent スキーマレジストリ URL
- （オプション）スキーマレジストリの URL（例：https://schema-registry.example.com:8081）。カスタムアクションでスキーマ検証を使用する場合に必要です。
Confluent スキーマレジストリユーザー名
- （オプション）HTTP Basic 認証を介してスキーマレジストリに認証するために使用されるユーザー名。スキーマレジストリが認証を強制する場合に必要です。
Confluent スキーマレジストリパスワード
- （オプション）HTTP Basic 認証を介してスキーマレジストリに認証するために使用されるパスワードまたは API シークレット。スキーマレジストリが認証を強制する場合に必要です。
サービスアカウントユーザー名
- （必須）SASL/PLAIN で認証するための Kafka プリンシパル。
サービスアカウントトークン / パスワード
- （必須）サービスアカウントプリンシパルの JWT トークンまたは同等の資格情報。
圧縮タイプ
- Kafka に送信されるメッセージの圧縮アルゴリズム。帯域幅と保存コストを削減します。空白の場合、GZIP が使用されます。
最大メッセージサイズ（バイト）
- 一回のリクエストで送信される単一メッセージまたはバッチの最大サイズ（バイト）。大きな訪問プロファイルを収容するためにこの値を増やしますが、ブローカーの構成と一致させる必要があります。デフォルト：1,048,576（1 MB）。最大：2,097,152（2 MB）。Kafka ブローカーの message.max.bytes 構成を超えてはいけません。
クライアントID
- ログ、モニタリングの帰属、およびクォータの強制のために、すべてのリクエストとともにブローカーに送信される文字列識別子。空白の場合、tealium-{account}-{profile}-{connectorId} の形式で自動生成された識別子が使用されます。許可される文字：文字、数字、ドット、ハイフン、アンダースコア。
確認（acks）
- メッセージの受領を確認する必要があるブローカーレプリカの数。1（リーダーのみ）が最も速い。All はすべての同期レプリカを待ちますが、50-100ms の遅延が発生し、データ損失のリスクがなくなります。0 は確認なしで、最も速いが、無音のデータ損失のリスクがあります。
パーティショナー戦略
- メッセージキーが構成されていない場合、プロデューサーがパーティションにメッセージを割り当てる方法を制御します。デフォルト は最適なスループットのためのスティッキーバッチングを使用します。ラウンドロビン はメッセージをすべてのパーティションに均等に分配します。メッセージキーが構成されている場合、この構成に関係なく、常にキーハッシュによってパーティション分けされます。
プロデューサー - 再接続バックオフ
- 接続失敗後にブローカーに再接続を試みる前に待つミリ秒数。空白の場合、Kafka クライアントのデフォルトの 50ms が使用されます。
プロデューサー - リトライ
- 送信操作が失敗した場合の最大リトライ回数。空白の場合、Kafka クライアントのデフォルトが使用されます。
プロデューサー - リトライバックオフ
- 同じレコードの連続する送信リトライの間に待つミリ秒数。空白の場合、Kafka クライアントのデフォルトの 100ms が使用されます。

アクション

アクション名	AudienceStream	EventStream
すべてのイベントデータを送信	✗	✓
すべての訪問データを送信	✓	✗
カスタムイベントデータを送信	✗	✓
カスタム訪問データを送信	✓	✗
すべてのログイベントを送信	✗	✓
ログイベントを送信	✗	✓

次へをクリックするか、アクション タブに進みます。ここでコネクタアクションを構成します。

次のセクションでは、各アクションのパラメータとオプションの構成方法について説明します。

すべてのイベントデータを送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力します。
データ形式	データ配信の形式を指定します：`JSON`、`STRING`、または `BINARY`。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、メッセージキーまたはラウンドロビンに基づいて Kafka のデフォルトパーティショナーが使用されます。
ヘッダー	（オプション）Kafka レコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式が `BINARY` に構成されている場合、この値は送信前にエンコードされます。
タイムスタンプ	（オプション）ISO 8601 UTC 形式のメッセージタイムスタンプ `YYYY-MM-DDThh:mm:ssZ`。提供されていない場合、現在のタイムスタンプが使用されます。
属性名の印刷	デフォルトでは、属性キーが使用されます。このオプションを有効にすると、キーとして属性名が使用されます。
バッチの有効期限	バッチの有効期限（分単位、1から60の間）。デフォルト：10。

すべての訪問データを送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力します。
データ形式	データ配信の形式を指定します：`JSON`、`STRING`、または `BINARY`。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、メッセージキーまたはラウンドロビンに基づいて Kafka のデフォルトパーティショナーが使用されます。
ヘッダー	（オプション）Kafka レコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式が `BINARY` に構成されている場合、Tealium はこの値をエンコードしてから送信します。
タイムスタンプ	（オプション）ISO 8601 UTC 形式のメッセージタイムスタンプ `YYYY-MM-DDThh:mm:ssZ`。提供されていない場合、現在のタイムスタンプが使用されます。
属性名の印刷	デフォルトでは、属性キーが使用されます。このオプションを有効にすると、キーとして属性名が使用されます。
訪問データを訪問データに含める	訪問データを訪問データペイロードに含めます。
バッチの有効期限	バッチの有効期限（分単位、1から60の間）。デフォルト：10。

カスタムイベントデータを送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力します。
データ形式	データ配信の形式を指定します：`JSON`、`STRING`、または `BINARY`。
スキーマ	（オプション）スキーマレジストリからスキーマ主題を選択します。スキーマレジストリ URL の構成が必要です。
メッセージデータ	メッセージデータを構築するための値を提供します。テンプレートを使用する場合は、テンプレート名を参照し、カスタムメッセージ定義にマップします。テンプレートが使用される場合、他のキーバリューペアは無視されます。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、メッセージキーまたはラウンドロビンに基づいて Kafka のデフォルトパーティショナーが使用されます。
ヘッダー	（オプション）Kafka レコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式が `BINARY` に構成されている場合、Tealium はこの値をエンコードしてから送信します。
タイムスタンプ	（オプション）ISO 8601 UTC 形式のメッセージタイムスタンプ `YYYY-MM-DDThh:mm:ssZ`。提供されていない場合、現在のタイムスタンプが使用されます。
バッチの有効期限	バッチの有効期限（分単位、1から60の間）。デフォルト：10。

カスタム訪問データを送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力します。
データ形式	データ配信の形式を指定します：`JSON`、`STRING`、または `BINARY`。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、メッセージキーまたはラウンドロビンに基づいて Kafka のデフォルトパーティショナーが使用されます。
ヘッダー	（オプション）Kafka レコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式が `BINARY` に構成されている場合、Tealium はこの値をエンコードしてから送信します。
タイムスタンプ	（オプション）ISO 8601 UTC 形式のメッセージタイムスタンプ `YYYY-MM-DDThh:mm:ssZ`。提供されていない場合、現在のタイムスタンプが使用されます。
属性名の印刷	デフォルトでは、属性キーが使用されます。このオプションを有

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力してください。
データ形式	データ配信の形式を指定します: `JSON`, `STRING`, `BINARY`。
スキーマ	（オプション）スキーマレジストリからスキーマ主題を選択します。スキーマレジストリのURLが構成されている必要があります。
メッセージデータ	メッセージデータを構築するための値を提供します。テンプレートを使用する場合は、テンプレート名を参照し、カスタムメッセージ定義にマップします。テンプレートが使用される場合、他のキー値ペアは無視されます。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、Kafkaのデフォルトパーティショナーがメッセージキーまたはラウンドロビンに基づいて使用されます。
ヘッダー	（オプション）Kafkaレコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式としてBINARYを選択した場合、Tealiumはこの値をエンコードして送信します。
タイムスタンプ	（オプション）メッセージのタイムスタンプをISO 8601 UTC形式 `YYYY-MM-DDThh:mm:ssZ` で指定します。提供されない場合、現在のタイムスタンプが使用されます。
バッチの有効時間	バッチの有効時間（分単位で1から60の間）。デフォルト: 10。

ログイベント全体を送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力してください。
データ形式	データ配信の形式を指定します: `JSON`, `STRING`, `BINARY`。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、Kafkaのデフォルトパーティショナーがメッセージキーまたはラウンドロビンに基づいて使用されます。
ヘッダー	（オプション）Kafkaレコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式としてBINARYを選択した場合、Tealiumはこの値をエンコードして送信します。
タイムスタンプ	（オプション）メッセージのタイムスタンプをISO 8601 UTC形式 `YYYY-MM-DDThh:mm:ssZ` で指定します。提供されない場合、現在のタイムスタンプが使用されます。
バッチの有効時間	バッチの有効時間（分単位で1から60の間）。デフォルト: 10。

ログイベントを送信

パラメータ

パラメータ	説明
トピック	トピックを選択するか、トピックIDを入力してください。
データ形式	データ配信の形式を指定します: `JSON`, `STRING`, `BINARY`。
パーティションID	（オプション）パーティションIDを選択します。空白の場合、Kafkaのデフォルトパーティショナーがメッセージキーまたはラウンドロビンに基づいて使用されます。
ヘッダー	（オプション）Kafkaレコードヘッダーに追加されるキーバリューペアのセット。
メッセージキー	（オプション）生のメッセージキーを指定します。データ形式としてBINARYを選択した場合、Tealiumはこの値をエンコードして送信します。
タイムスタンプ	（オプション）メッセージのタイムスタンプをISO 8601 UTC形式 `YYYY-MM-DDThh:mm:ssZ` で指定します。提供されない場合、現在のタイムスタンプが使用されます。
バッチの有効時間	バッチの有効時間（分単位で1から60の間）。デフォルト: 10。

コネクタログパラメータ

パラメータの説明については、コネクタログパラメータを参照してください。