` 配下) です。最初にグローバル設定が適用され、その後、トピックレベルの設定が存在する場合はそれが適用されます。
```xml theme={null}
cgrp
3000
logs
4000
smallest
logs
100000
stats
50000
logs
250
stats
400
```
利用可能な設定オプションの一覧については、[librdkafka の設定リファレンス](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)を参照してください。ClickHouse の設定では、ドットの代わりにアンダースコア (`_`) を使用してください。たとえば、`check.crcs=true` は `true` となります。
### Kerberos サポート
Kerberos 対応の Kafka を扱うには、値に `sasl_plaintext` を指定した `security_protocol` 子要素を追加します。Kerberos のチケット授与チケットが OS の機能によって取得され、キャッシュされていれば十分です。
ClickHouse は、keytab ファイルを使用して Kerberos 認証情報を保持できます。`sasl_kerberos_service_name`、`sasl_kerberos_keytab`、`sasl_kerberos_principal` の各子要素も指定できます。
例:
```xml theme={null}
SASL_PLAINTEXT
/home/kafkauser/kafkauser.keytab
kafkauser/kafkahost@EXAMPLE.COM
```
## 仮想カラム
* `_topic` — Kafkaトピック。データ型: `LowCardinality(String)`.
* `_key` — メッセージのキー。データ型: `String`.
* `_offset` — メッセージのオフセット。データ型: `UInt64`.
* `_timestamp` — メッセージのタイムスタンプ。データ型: `Nullable(DateTime)`.
* `_timestamp_ms` — メッセージのミリ秒単位のタイムスタンプ。データ型: `Nullable(DateTime64(3))`.
* `_partition` — Kafkaトピックのパーティション。データ型: `UInt64`.
* `_headers.name` — メッセージヘッダーのキーの Array。データ型: `Array(String)`.
* `_headers.value` — メッセージヘッダーの値の Array。データ型: `Array(String)`.
`kafka_handle_error_mode='stream'` の場合に追加される仮想カラム:
* `_raw_message` - 正常にパースできなかった生のメッセージ。データ型: `String`.
* `_error` - パース失敗時に発生した例外メッセージ。データ型: `String`.
注: 仮想カラム `_raw_message` と `_error` に値が入るのは、パース中に例外が発生した場合のみです。メッセージが正常にパースされた場合、これらは常に空です。
## Kafka メッセージのメタデータへのカラムのマッピング
`INSERT INTO` でメッセージを生成するとき、Kafka エンジンは、テーブルにそれらのカラムが存在する場合、常に `_key` という名前のカラム (型は `String`) を Kafka メッセージキーとして、`_timestamp` という名前のカラム (型は `DateTime`) を Kafka メッセージの timestamp として使用します。デフォルトでは、これらのカラムは生成されるメッセージの payload にも、ほかのカラムとあわせて含まれます。
`kafka_map_virtual_columns_on_write = 1` を使用すると、動作は次のように変わります。
* `_key` (型 `String`) — Kafka メッセージキーにマッピングされます。
* `_timestamp` (型 `DateTime`) — Kafka メッセージの timestamp にマッピングされます。
* `_headers.name` (型 `Array(String)`) および `_headers.value` (型 `Array(String)`) — Kafka メッセージヘッダーにマッピングされます。各ペア `(_headers.name[i], _headers.value[i])` は 1 つの Kafka ヘッダーになります。`_headers.name` と `_headers.value` は `_headers` という Nested プレフィックスを共有しているため、ClickHouse ではすべての行で両方の配列のサイズが同じでなければなりません。
これらの名前を持つカラムは、型が上記のものと一致する場合にのみ **メッセージの payload から除外されます**。それ以外の場合は payload に残るため、たまたまこれらの名前を無関係なデータに使っているスキーマでも引き続き動作します。
例:
```sql theme={null}
CREATE TABLE kafka_out
(
event_json String,
`_key` String,
`_timestamp` DateTime,
`_headers.name` Array(String),
`_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'events',
kafka_group_name = 'events-producer',
kafka_format = 'JSONEachRow',
kafka_map_virtual_columns_on_write = 1;
INSERT INTO kafka_out VALUES
('{"a":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);
```
生成された Kafka メッセージには、ペイロード `{"event_json":"{\"a\":1}"}`、キー `session-42`、現在のタイムスタンプ、および 2 つのヘッダー `source=api` と `trace_id=abc-123` が含まれます。
## データフォーマットのサポート
Kafka エンジンは、ClickHouse でサポートされているすべての[フォーマット](/ja/reference/formats)に対応しています。
1 つの Kafka メッセージに含まれる行数は、そのフォーマットが行ベースかブロックベースかによって異なります。
* 行ベースのフォーマットでは、1 つの Kafka メッセージに含める行数を `kafka_max_rows_per_message` の設定で制御できます。
* ブロックベースのフォーマットでは、ブロックをさらに小さな部分に分割することはできませんが、1 つのブロックに含まれる行数は一般設定の [max\_block\_size](/ja/reference/settings/session-settings#max_block_size) で制御できます。
## ClickHouse Keeper にコミット済みオフセットを保存するエンジン
`allow_experimental_kafka_offsets_storage_in_keeper` が有効な場合、Kafka テーブルエンジンにはさらに 2 つの設定を指定できます。
* `kafka_keeper_path` は ClickHouse Keeper 内のテーブルへのパスを指定します
* `kafka_replica_name` は ClickHouse Keeper 内のレプリカ名を指定します
これらの設定は、両方を指定するか、どちらも指定しないようにする必要があります。両方を指定すると、新しい実験的な Kafka エンジンが使用されます。この新しいエンジンは、コミット済みオフセットを Kafka に保存することに依存せず、ClickHouse Keeper に保存します。引き続き Kafka へのオフセットの commit は試みますが、それらのオフセットに依存するのはテーブルの作成時だけです。それ以外の状況 (テーブルの再起動時や、何らかの error からの復旧後など) では、ClickHouse Keeper に保存されたオフセットが、メッセージ消費を継続するためのオフセットとして使用されます。コミット済みオフセットに加えて、直前のバッチで何件のメッセージを消費したかも保存されるため、insert が失敗した場合でも同じ件数のメッセージが消費され、必要に応じて重複排除を行えます。
例:
```sql theme={null}
CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;
```
### 既知の制限事項
この新しいエンジンは実験的機能のため、まだ本番環境での利用には適していません。実装には、いくつか既知の制限があります。
* テーブルを短時間で削除して再作成したり、同じ ClickHouse Keeper パスを異なるエンジンに指定したりすると、問題が発生する可能性があります。ベストプラクティスとして、パスの衝突を避けるために `kafka_keeper_path` で `{uuid}` を使用できます。
* 再現可能な読み取りを実現するには、1 つのスレッドで複数のパーティションからメッセージを消費することはできません。一方で、Kafka コンシューマーを維持するには、定期的にポーリングする必要があります。これら 2 つの要件があるため、複数のコンシューマーを作成できるのは `kafka_thread_per_consumer` が有効な場合のみにしています。そうでない場合、コンシューマーを定期的にポーリングしながら問題を避けるのが複雑になりすぎるためです。
**関連項目**
* [仮想カラム](/ja/reference/engines/table-engines#table_engines-virtual_columns)
* [background\_message\_broker\_schedule\_pool\_size](/ja/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size)
* [system.kafka\_consumers](/ja/reference/system-tables/kafka_consumers)