` | MAP | Map(K, V) | ✅ | |
| oneof | STRUCT | Tuple / Variant | ⚠️ | oneof を ClickHouse スキーマに変換する方法については、以下のセクションを参照してください |
| google.protobuf.DoubleValue | FLOAT64 | Nullable(Float64) | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | Nullable(Float32) | ✅ | |
| google.protobuf.Int64Value | INT64 | Nullable(Int64) | ✅ | |
| google.protobuf.UInt64Value | INT64 | Nullable(UInt64) | ✅ | |
| google.protobuf.UInt32Value | INT64 | Nullable(UInt32) | ✅ | |
| google.protobuf.Int32Value | INT32 | Nullable(Int32) | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | Nullable(Bool) | ✅ | |
| google.protobuf.StringValue | STRING | Nullable(String) | ✅ | |
| google.protobuf.BytesValue | BYTES | Nullable(String) | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | DateTime64(3) | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | Int32 / Int64 | ✅ | |
| google.protobuf.Duration | STRUCT | Tuple(`seconds` Int64, `nano` Nullable(Int32)) | ✅ | |
| google.protobuf.Any | *N/A* | *N/A* | ❌ | |
| google.protobuf.Empty | *N/A* | *N/A* | ❌ | |
Kafka Connect の型と ClickHouse の型マッピングについては、[サポートされているデータ型](#supported-data-types)を参照してください。
#### `oneof` フィールドを ClickHouse のカラムに変換する際の注意
このコネクタは、Protobuf のユニオン (`oneof`) を ClickHouse の Variant 型に変換することをサポートしていません。代わりに、ClickHouse テーブルのスキーマでは、`oneof` フィールドを個別の Nullable フィールドとして列挙してください。
たとえば:
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message StringIntUnion {
oneof mixed {
string mixed_string = 2;
int32 mixed_int = 3;
}
}
```
次の ClickHouse テーブル定義に変換されます:
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
#### サポートされていない Protobuf スキーマ
コネクタでは、以下の Protobuf スキーマはサポートされていません。
* 複数メッセージのユニオン (**CH バージョン 26.1 より前**)
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message TwoRecords {
oneof payload {
TypeA type_a = 2;
TypeB type_b = 3;
}
// translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
message TypeA {
string label = 1;
}
// translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
message TypeB {
int32 count = 1;
}
}
```
CH バージョン 26.1 以降では、`allow_experimental_nullable_tuple_type=1` を設定すると、このスキーマがサポートされます ([こちらのドキュメントページ](/docs/ja/reference/settings/session-settings#allow_experimental_nullable_tuple_type)を参照) 。
### JSON スキーマ対応
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
### Stringのサポート
このコネクタは、ClickHouse のさまざまなフォーマットで String Converter をサポートしています：[JSON](/docs/ja/reference/formats/JSON/JSONEachRow)、[CSV](/docs/ja/reference/formats/CSV/CSV)、および [TSV](/docs/ja/reference/formats/TabSeparated/TabSeparated)。
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"customInsertFormat": "true",
"insertFormat": "CSV"
}
}
```
### 内部バッファリング
内部バッファリングを使用すると、シンクタスクは複数回の `poll()` 呼び出しで取得したレコードを蓄積し、より大きなバッチとして ClickHouse にフラッシュできます。これにより、各 `poll` でパーティションごとの小さなバッチが多数生成されるワークロードでは、スループットが向上する場合があります。
主な動作:
* `bufferCount` は、フラッシュ前にバッファリングするレコード数を制御します。
* `bufferFlushTime` は、バッファリングされたレコードをフラッシュするまでの最大待機時間 (ミリ秒) を設定します。
* `bufferFlushTime` は、`bufferCount > 0` の場合にのみ有効です。
* `bufferCount=0` および `bufferFlushTime=0` の場合、バッファリングは無効のままです (デフォルトの動作) 。
* `exactlyOnce=true` の場合、バッファリングはサポートされません。
バッファリングが exactly-once モードと互換性がない理由:
バッファリングによってバッチ境界が変わるため、ClickHouse のブロック重複排除と、コネクタのオフセット状態マシンが正しく機能しなくなります。
これを回避するには、コネクタ設定で `exactlyOnce=false` を指定して exactly-once モードを無効にするか、`bufferCount=0` を指定してバッファリングを無効にしてください。
例:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### ロギング
ロギングは Kafka Connect Platform で自動的に提供されます。
ログの宛先とフォーマットは、Kafka Connect の[設定ファイル](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file)で設定できます。
Confluent Platform を使用している場合は、CLI コマンドを実行してログを確認できます。
```bash theme={null}
confluent local services connect log
```
詳細については、公式の[チュートリアル](https://docs.confluent.io/platform/current/connect/logging.html)を参照してください。
### 監視
ClickHouse Kafka Connect は、[Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html) を通じてランタイムメトリクスを公開します。JMX は Kafka Connector でデフォルトで有効化されています。
#### ClickHouse固有のメトリクス
このコネクタは、次のMBean名でカスタムメトリクスを公開します。
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| メトリック名 | 型 | 説明 |
| ---------------------- | ---- | --------------------------------------- |
| `receivedRecords` | long | 受信したレコードの総数。 |
| `recordProcessingTime` | long | レコードのグループ化と統一された構造への変換に要した合計時間 (ナノ秒) 。 |
| `taskProcessingTime` | long | データの処理と ClickHouse への挿入に要した合計時間 (ナノ秒) 。 |
#### Kafka Producer/Consumer メトリクス
このコネクタは、データフロー、スループット、パフォーマンスの把握に役立つ、Kafka の標準的なプロデューサー/コンシューマーメトリクスを公開します。
**トピックレベルのメトリクス:**
* `records-sent-total`: トピックに送信されたレコードの総数
* `bytes-sent-total`: トピックに送信された総バイト数
* `record-send-rate`: 1 秒あたりに送信されたレコードの平均レート
* `byte-rate`: 1 秒あたりに送信された平均バイト数
* `compression-rate`: 達成された圧縮率
**パーティションレベルのメトリクス:**
* `records-sent-total`: パーティションに送信されたレコードの総数
* `bytes-sent-total`: パーティションに送信された総バイト数
* `records-lag`: パーティションの現在のラグ
* `records-lead`: パーティションの現在のリード
* `replica-fetch-lag`: レプリカのラグ情報
**ノードレベルの接続メトリクス:**
* `connection-creation-total`: Kafka ノードに対して作成された接続の総数
* `connection-close-total`: クローズされた接続の総数
* `request-total`: ノードに送信されたリクエストの総数
* `response-total`: ノードから受信したレスポンスの総数
* `request-rate`: 1 秒あたりの平均リクエストレート
* `response-rate`: 1 秒あたりの平均レスポンスレート
これらのメトリクスは、次の点の監視に役立ちます。
* **スループット**: データのインジェスト率を追跡
* **ラグ**: ボトルネックや処理遅延を特定
* **圧縮**: データ圧縮の効率を測定
* **接続の健全性**: ネットワーク接続の状態と安定性を監視
#### Kafka Connect Framework のメトリクス
このコネクタは Kafka Connect Framework と統合されており、タスクのライフサイクルとエラー追跡に関するメトリクスを公開します。
**タスクステータスのメトリクス:**
* `task-count`: コネクタ内のタスク総数
* `running-task-count`: 現在実行中のタスク数
* `paused-task-count`: 現在一時停止中のタスク数
* `failed-task-count`: 失敗したタスク数
* `destroyed-task-count`: 破棄されたタスク数
* `unassigned-task-count`: 未割り当てのタスク数
タスクステータスの値には、`running`、`paused`、`failed`、`destroyed`、`unassigned` があります
**エラーメトリクス:**
* `deadletterqueue-produce-failures`: 失敗した DLQ への書き込み数
* `deadletterqueue-produce-requests`: DLQ への書き込み試行総数
* `last-error-timestamp`: 直近のエラーのタイムスタンプ
* `records-skip-total`: エラーによりスキップされたレコード総数
* `records-retry-total`: 再試行されたレコード総数
* `errors-total`: 発生したエラーの総数
**パフォーマンスメトリクス:**
* `offset-commit-failures`: オフセットコミットの失敗数
* `offset-commit-avg-time-ms`: オフセットコミットの平均所要時間
* `offset-commit-max-time-ms`: オフセットコミットの最大所要時間
* `put-batch-avg-time-ms`: バッチ処理の平均所要時間
* `put-batch-max-time-ms`: バッチ処理の最大所要時間
* `source-record-poll-total`: ポーリングされたレコード総数
#### 監視のベストプラクティス
1. **コンシューマラグを監視する**: 処理のボトルネックを特定するため、パーティションごとの `records-lag` を追跡します
2. **エラー率を追跡する**: データ品質の問題を検出するため、`errors-total` と `records-skip-total` を監視します
3. **タスクの健全性を確認する**: タスクが正常に実行されていることを確認するため、タスクのステータスメトリクスを監視します
4. **スループットを測定する**: インジェストのパフォーマンスを追跡するため、`records-send-rate` と `byte-rate` を使用します
5. **接続の健全性を監視する**: ネットワークの問題を確認するため、ノードレベルの接続メトリクスを確認します
6. **圧縮効率を追跡する**: データ転送を最適化するため、`compression-rate` を使用します
JMX メトリクスの詳細な定義と Prometheus インテグレーションについては、[jmx-export-connector.yml](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml) 設定ファイルを参照してください。
### 制限事項
* 削除には対応していません。
* バッチサイズは Kafka Consumer のプロパティを継承します。
* exactly-once のために KeeperMap を使用している場合、オフセットを変更または巻き戻したときは、その トピック に対応する KeeperMap の内容を削除する必要があります。 (詳細は下記のトラブルシューティングガイドを参照してください)
### パフォーマンスチューニングとスループット最適化
このセクションでは、ClickHouse Kafka Connect Sink のパフォーマンスチューニングの方法について説明します。パフォーマンスチューニングは、高スループットのユースケースに対応する場合や、リソース使用率を最適化して遅延を最小限に抑える必要がある場合に不可欠です。
#### パフォーマンスチューニングが必要になるのはどのような場合ですか？
パフォーマンスチューニングは、通常、次のような状況で必要になります。
* **高スループットのワークロード**: Kafkaトピックから毎秒数百万件のイベントを処理する場合
* **コンシューマラグ**: データ生成の速度にコネクタが追いつかず、ラグが増大している場合
* **リソース制約**: CPU、メモリ、またはネットワークの使用を最適化する必要がある場合
* **複数のトピック**: 大量のデータが流れる複数のトピックを同時に消費する場合
* **小さなメッセージサイズ**: サーバー側のバッチ処理の恩恵を受けられる、小さなメッセージを大量に扱う場合
パフォーマンスチューニングが**通常は不要**なのは、次のような場合です。
* 低〜中程度の量 (毎秒10,000メッセージ未満) を処理している
* コンシューマラグが安定しており、ユースケース上許容できる範囲に収まっている
* デフォルトのコネクタ設定ですでに必要なスループット要件を満たしている
* ClickHouseクラスターが流入する負荷を容易に処理できる
#### データフローを理解する
チューニングを行う前に、データがコネクタ内をどのように流れるかを理解しておくことが重要です。
1. **Kafka Connect Framework** がバックグラウンドで Kafka トピックからメッセージを取得します
2. **コネクタ** がフレームワークの内部バッファからメッセージをポーリングします
3. **コネクタ** はポーリングサイズに基づいてメッセージをバッチ化します
4. **ClickHouse** は HTTP/S 経由でバッチ化された insert を受け取ります
5. **ClickHouse** はその insert を処理します (同期または非同期)
これらの各段階でパフォーマンスを最適化できます。
#### Kafka Connect のバッチサイズ調整
最初の最適化ポイントは、コネクタが Kafka から 1 バッチごとに受け取るデータ量を制御することです。
##### fetch設定
Kafka Connect (フレームワーク) は、コネクタとは独立して、バックグラウンドで Kafka トピックからメッセージをfetchします。
* **`fetch.min.bytes`**: フレームワークがデータをコネクタに渡す前に必要な最小データ量 (デフォルト: 1 byte)
* **`fetch.max.bytes`**: 1 回のリクエストでfetchできるデータ量の上限 (デフォルト: 52428800 / 50 MB)
* **`fetch.max.wait.ms`**: `fetch.min.bytes` に達しない場合に、データを返すまでの最大待機時間 (デフォルト: 500 ms)
Confluent Cloud では、これらの設定を変更するには Confluent Cloud でサポートケースを作成する必要があります。
##### ポーリング設定
コネクタはフレームワークのバッファからメッセージをポーリングします。
* **`max.poll.records`**: 1 回のポーリングで返されるレコードの最大数 (デフォルト: 500)
* **`max.partition.fetch.bytes`**: パーティションごとの最大データ量 (デフォルト: 1048576 / 1 MB)
Confluent Cloud では、これらの設定を変更するには、Confluent Cloud のサポートケースを作成する必要があります。
##### 高スループット向けの推奨設定
ClickHouseで最適なパフォーマンスを得るには、より大きなバッチを使用するようにしてください。
```properties theme={null}
# ポーリングごとのレコード数を増やす
consumer.override.max.poll.records=5000
# パーティションのフェッチサイズを増やす（5 MB）
consumer.override.max.partition.fetch.bytes=5242880
# オプション: より多くのデータを待つために最小フェッチサイズを増やす（1 MB）
consumer.override.fetch.min.bytes=1048576
# オプション: レイテンシが重要な場合は待機時間を短縮する
consumer.override.fetch.max.wait.ms=300
```
上記のプロパティを使用するには、`connector.client.config.override.policy=All` を指定して、ワーカー設定でクライアントのオーバーライドを有効にする必要があります。詳細は、[Kafka Connect ドキュメント](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration)を参照してください。
**重要**: Kafka Connect の fetch 設定は圧縮データを表しますが、ClickHouse が受信するのは非圧縮データです。これらの設定は、圧縮率を踏まえて調整してください。
**トレードオフ**:
* **バッチが大きいほど** = ClickHouse へのインジェスト性能が向上し、パーツ数が減り、オーバーヘッドが低減
* **バッチが大きいほど** = メモリ使用量が増え、エンドツーエンドのレイテンシが高くなる可能性
* **バッチが大きすぎる場合** = タイムアウト、OutOfMemory エラー、または `max.poll.interval.ms` 超過のリスク
詳細: [Confluent ドキュメント](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [Kafka ドキュメント](https://kafka.apache.org/documentation/#consumerconfigs)
#### 非同期挿入
非同期挿入は、コネクタが比較的小さなバッチを送信する場合や、バッチ処理を ClickHouse に任せてインジェストをさらに最適化したい場合に有効な強力な機能です。
##### 非同期 INSERT を使用するタイミング
次のような場合は、非同期 INSERT を有効にすることを検討してください。
* **小さなバッチが多数ある**: コネクタが小さなバッチ (1 バッチあたり 1000 行未満) を高頻度で送信している
* **高い並行性**: 複数のコネクタ タスクが同じテーブルに書き込んでいる
* **分散デプロイ**: 異なるホスト上で多数のコネクタ インスタンスを実行している
* **パーツ作成のオーバーヘッド**: 「パーツが多すぎる」エラーが発生している
* **混在ワークロード**: リアルタイム インジェストとクエリ ワークロードを組み合わせている
次のような場合は、非同期 INSERT を使用しないでください。
* すでに大きなバッチ (1 バッチあたり 10,000 行超) を、頻度を制御しながら送信している
* データを即座に可視化する必要がある (クエリですぐにデータを参照できる必要がある)
* `wait_for_async_insert=0` を使用する exactly-once セマンティクス が要件と競合する
* 代わりにクライアント側のバッチ処理の改善で効果を得られるユースケースである
##### 非同期 INSERT の仕組み
非同期挿入を有効にすると、ClickHouse は次のように動作します。
1. コネクタから INSERT クエリを受け取ります
2. データをメモリ上のバッファに書き込みます (すぐにディスクへは書き込みません)
3. コネクタに成功を返します (`wait_for_async_insert=0` の場合)
4. 次のいずれかの条件が満たされると、バッファをディスクにフラッシュします。
* バッファが `async_insert_max_data_size` に達する (デフォルト: 100 MB)
* 最初の INSERT から `async_insert_busy_timeout_ms` ミリ秒が経過する (デフォルト: 1000 ms)
* 累積されたクエリ数が上限に達する (`async_insert_max_query_number`、デフォルト: 100)
これにより、作成されるパーツ数が大幅に減り、全体のスループットが向上します。
##### 非同期 INSERT を有効にする
`clickhouseSettings` 設定パラメータに、非同期 INSERT の設定を追加します。
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**主な設定**:
* **`async_insert=1`**: 非同期挿入を有効にします
* **`wait_for_async_insert=1`** (推奨) : コネクタは、データが ClickHouse ストレージにフラッシュされるまで待機してから確認応答します。これにより、配信が保証されます。
* **`wait_for_async_insert=0`**: コネクタは、バッファリング直後にただちに確認応答します。パフォーマンスは向上しますが、フラッシュ前にサーバーがクラッシュするとデータが失われる可能性があります。
##### 非同期 INSERT の動作の調整
非同期 INSERT のフラッシュ動作は細かく調整できます：
```json theme={null}
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"
```
一般的なチューニングパラメーター:
* **`async_insert_max_data_size`** (デフォルト: 104857600 / 100 MB): フラッシュ前の最大バッファサイズ
* **`async_insert_busy_timeout_ms`** (デフォルト: 1000): フラッシュまでの最大時間 (ms)
* **`async_insert_stale_timeout_ms`** (デフォルト: 0): 最後の insert からフラッシュまでの時間 (ms)
* **`async_insert_max_query_number`** (デフォルト: 100): フラッシュ前の最大クエリ数
**トレードオフ**:
* **利点**: パーツ数の削減、マージ性能の向上、CPU オーバーヘッドの低減、高い同時実行数下でのスループット向上
* **考慮事項**: データをすぐにはクエリできないこと、エンドツーエンドのレイテンシがわずかに増加すること
* **リスク**: `wait_for_async_insert=0` の場合、サーバークラッシュ時にデータが失われる可能性があること、大きなバッファによってメモリが逼迫する可能性があること
##### exactly-once セマンティクスの非同期 INSERT
非同期 INSERT で `exactlyOnce=true` を使用する場合:
```json theme={null}
{
"config": {
"exactlyOnce": "true",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**重要**: offset のコミットがデータの永続化後にのみ行われるようにするため、exactly-once では必ず `wait_for_async_insert=1` を使用してください。
非同期 INSERT の詳細については、[ClickHouse async inserts documentation](/docs/ja/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts) を参照してください。
#### コネクタの並列度
スループットを向上させるには、並列度を上げます。
##### コネクタあたりのタスク数
```json theme={null}
"tasks.max": "4"
```
各タスクは、トピックパーティションの一部を処理します。タスク数を増やすほど並列性は高まりますが、次の点に注意してください。
* 実質的に有効なタスク数の上限 = トピックパーティション数
* 各タスクは ClickHouse への接続をそれぞれ維持します
* タスク数が増えるほどオーバーヘッドが大きくなり、リソース競合が発生する可能性があります
**推奨**: まず `tasks.max` をトピックパーティション数と同じ値に設定し、その後 CPU 使用率とスループットのメトリクスを見ながら調整してください。
##### バッチ処理時にパーティションを区別しない
デフォルトでは、コネクタはパーティションごとにメッセージをバッチ処理します。スループットを高めるには、パーティションをまたいでバッチ処理できます。
```json theme={null}
"ignorePartitionsWhenBatching": "true"
```
\*\* 警告\*\*: `exactlyOnce=false` の場合にのみ使用してください。この設定では、より大きなバッチを作成してスループットを向上できますが、パーティションごとの順序保証は失われます。
#### 複数の高スループットなトピック
コネクタが複数のトピックを購読するように設定されており、`topic2TableMap` を使ってトピックをテーブルにマッピングしていて、挿入時のボトルネックによってコンシューマラグが発生している場合は、代わりにトピックごとにコネクタを 1 つずつ作成することを検討してください。
これが発生する主な理由は、現時点ではバッチが各テーブルに[直列で](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100)挿入されるためです。
**推奨事項**: 高ボリュームのトピックが複数ある場合は、並列挿入スループットを最大化するために、トピックごとにコネクタインスタンスを 1 つずつデプロイしてください。
#### ClickHouse テーブルエンジンに関する考慮事項
ユースケースに適した ClickHouse テーブルエンジンを選択してください。
* **`MergeTree`**: ほとんどのユースケースに最適で、クエリと insert のパフォーマンスのバランスに優れています
* **`ReplicatedMergeTree`**: 高可用性に必要ですが、レプリケーションのオーバーヘッドが増えます
* **適切な `ORDER BY` を設定した `*MergeTree`**: クエリパターンに合わせて最適化できます
**検討すべき設定**:
```sql theme={null}
CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS
-- 並列パート書き込みのために挿入スレッドの最大数を増やす
max_insert_threads = 4,
-- 信頼性向上のためにクォーラムを使用した挿入を許可する（ReplicatedMergeTree）
insert_quorum = 2
```
コネクタレベルの insert 設定:
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### 接続プーリングとタイムアウト
コネクタは ClickHouse への HTTP 接続を維持します。遅延の大きいネットワークでは、タイムアウトを調整してください。
```json theme={null}
"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"
```
* **`socket_timeout`** (デフォルト: 30000 ms) : 読み取り操作の最大待機時間
* **`connection_timeout`** (デフォルト: 10000 ms) : 接続確立までの最大待機時間
大きなバッチでタイムアウトエラーが発生する場合は、これらの値を増やしてください。
#### パフォーマンスの監視とトラブルシューティング
以下の主要なメトリクスを監視します。
1. **コンシューマラグ**: Kafka の監視ツールを使用して、パーティションごとのラグを追跡します
2. **コネクタのメトリクス**: JMX 経由で `receivedRecords`、`recordProcessingTime`、`taskProcessingTime` を監視します ([監視](#monitoring)を参照)
3. **ClickHouse のメトリクス**:
* `system.asynchronous_inserts`: 非同期 INSERT バッファの使用状況を監視します
* `system.parts`: マージの問題を検出するためにパーツ数を監視します
* `system.merges`: 実行中のマージを監視します
* `system.events`: `InsertedRows`、`InsertedBytes`、`FailedInsertQuery` を追跡します
**一般的なパフォーマンスの問題**:
| 症状 | 考えられる原因 | 解決策 |
| --------------- | ---------------------- | --------------------------------------------------- |
| コンシューマラグが大きい | バッチが小さすぎる | `max.poll.records` を増やし、非同期 INSERT を有効にする |
| "パーツが多すぎる" エラー | 小さな insert が頻繁に発生している | 非同期 INSERT を有効にし、バッチサイズを増やす |
| タイムアウトエラー | バッチサイズが大きすぎる、ネットワークが遅い | バッチサイズを減らし、`socket_timeout` を増やし、ネットワークを確認する |
| CPU 使用率が高い | 小さなパーツが多すぎる | 非同期 INSERT を有効にし、マージ関連の設定を調整する |
| OutOfMemory エラー | バッチサイズが大きすぎる | `max.poll.records`、`max.partition.fetch.bytes` を減らす |
| タスク負荷に偏りがある | パーティションの分散に偏りがある | パーティションをリバランスするか、`tasks.max` を調整する |
#### ベストプラクティスの要約
1. **まずはデフォルト設定で始め**、実際のパフォーマンスを測定してから調整する
2. **より大きなバッチを優先する**: 可能であれば、1 回の insert あたり 10,000～100,000 行を目安にする
3. **非同期 INSERT を使用する**: 小さなバッチを多数送信する場合や、高い同時実行性が求められる場合に使用する
4. **exactly-once セマンティクスでは常に `wait_for_async_insert=1` を使用する**
5. **水平スケールする**: `tasks.max` をパーティション数まで増やす
6. **スループットの高い トピック ごとに 1 つのコネクタを使用する**: スループットを最大化するため
7. **継続的に監視する**: コンシューマラグ、パーツ数、merge アクティビティを追跡する
8. **十分にテストする**: 本番環境にデプロイする前に、現実的な負荷をかけて設定変更を必ずテストする
#### 例: 高スループット構成
以下に、高スループット向けに最適化した完全な例を示します。
```json theme={null}
{
"name": "clickhouse-high-throughput",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
"tasks.max": "8",
"topics": "high_volume_topic",
"hostname": "my-clickhouse-host.cloud",
"port": "8443",
"database": "default",
"username": "default",
"password": "",
"ssl": "true",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"exactlyOnce": "false",
"ignorePartitionsWhenBatching": "true",
"consumer.override.max.poll.records": "10000",
"consumer.override.max.partition.fetch.bytes": "5242880",
"consumer.override.fetch.min.bytes": "1048576",
"consumer.override.fetch.max.wait.ms": "500",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
}
}
```
上記のコネクタ設定では、worker の設定で `connector.client.config.override.policy=All` を指定し、クライアント設定のオーバーライドを有効にする必要があります。詳しくは、[Kafka Connect のドキュメント](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration)を参照してください。
**この設定**:
* 1 回の poll で最大 10,000 件のレコードを処理します
* より大きな INSERT を行うため、複数のパーティションにまたがってバッチ化します
* 16 MB のバッファで非同期 INSERT を使用します
* 8 個の task を並列実行します (パーティション数に合わせてください)
* 厳密な順序性よりもスループットを重視して最適化されています
### トラブルシューティング
#### "トピック `[someTopic]` パーティション `[0]` の状態不一致"
これは、KeeperMap に保存されているオフセットと Kafka に保存されているオフセットが一致しない場合に発生します。通常は、トピックが削除されたか
オフセットが手動で調整された場合です。
これを修正するには、該当するトピックとパーティションに保存されている古い値を削除する必要があります:
```sql theme={null}
-- まず、データの保存に使用されているデータベースを特定します。
SELECT * FROM [database].connect_state
-- トピックとパーティションに一致するキーを特定します。
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]
```
この調整は exactly-once に影響する可能性があります。
#### "コネクタはどのようなエラーで再試行しますか？"
現在は、一時的で再試行可能なエラーの特定に重点を置いており、以下が含まれます。
* `ClickHouseException` - これは ClickHouse によってスローされる汎用的な例外です。
通常はサーバーが過負荷のときにスローされ、特に一時的なものと見なされるエラーコードは次のとおりです。
* 3 - UNEXPECTED\_END\_OF\_FILE
* 107 - FILE\_DOESNT\_EXIST
* 159 - TIMEOUT\_EXCEEDED
* 164 - READONLY
* 202 - TOO\_MANY\_SIMULTANEOUS\_QUERIES
* 203 - NO\_FREE\_CONNECTION
* 209 - SOCKET\_TIMEOUT
* 210 - NETWORK\_ERROR
* 241 - MEMORY\_LIMIT\_EXCEEDED
* 242 - TABLE\_IS\_READ\_ONLY
* 252 - TOO\_MANY\_PARTS
* 285 - TOO\_FEW\_LIVE\_REPLICAS
* 319 - UNKNOWN\_STATUS\_OF\_INSERT
* 425 - SYSTEM\_ERROR
* 999 - KEEPER\_EXCEPTION
* `SocketTimeoutException` - これはソケットがタイムアウトしたときにスローされます。
* `UnknownHostException` - これはホスト名を解決できないときにスローされます。
* `IOException` - これはネットワークに問題があるときにスローされます。
#### "すべてのデータが空白/ゼロになっている"
おそらく、データ内のフィールドがテーブル内のフィールドと一致していません。これは特に CDC (変更データキャプチャ) や Debezium フォーマットでよく発生します。
一般的な解決策の 1 つは、コネクタ設定に flatten transformation を追加することです。
```properties theme={null}
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```
これにより、データはネストされた JSON からフラット化された JSON に変換されます (区切り文字として `_` を使用) 。その結果、テーブル内のフィールドは "field1\_field2\_field3" という形式 (つまり "before\_id"、"after\_id" など) に従います。
#### "ClickHouse で Kafka のキーを使いたい"
Kafka のキーはデフォルトでは `value` フィールドに保存されませんが、`KeyToValue` 変換を使うと、キーを `value` フィールド内の新しい `_key` フィールドに移動できます。
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```
