---
title: Apache Kafka コネクタ構成ガイド
description: この記事では、Customer Data Hub アカウントで Apache Kafka コネクタを構成する方法について説明します。
url: https://docs.tealium.com/ja/server-side-connectors/apache-kafka-connector/
---
## 構成

コネクタマーケットプレイスに移動し、新しいコネクタを追加します。コネクタを追加する一般的な手順については、[コネクタについて](https://docs.tealium.com/about-connectors/)を参照してください。

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

* **ブートストラップサーバー**
  * （必須）エンドポイント用の 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。 |

#### コネクタログパラメータ

パラメータの説明については、[コネクタログパラメータ](https://docs.tealium.com/connector-error-logging/#connector-log-parameters)を参照してください。