ClickHouse Kafka Connect Sink

ClickHouse Kafka Connect Sink は、Kafka トピックから ClickHouse テーブルにデータを取り込む Kafka コネクタです。

Kafka Connector Sink は、Apache 2.0 ライセンス に基づいて配布されています

環境に Kafka Connect フレームワーク v2.7 以降がインストールされ、Java 11 以降で実行されている必要があります。

このコネクタは、23.3 より古い ClickHouse サーバーでは起動しません。古いバージョンに固定する理由がない限り、常に最新リリースを使用してください。

• すぐに使える exactly-once セマンティクスを備えています。これは、KeeperMap という ClickHouse の新しいコア機能 (コネクタの状態ストアとして使用) を基盤としており、シンプルなアーキテクチャを実現します。
• 3rd-party の状態ストアをサポート: 現在のデフォルトはインメモリですが、KeeperMap も利用できます (Redis は近日追加予定) 。
• コアインテグレーション: ClickHouse により開発・保守・サポートされています。
• ClickHouse Cloud に対して継続的にテストされています。
• スキーマを定義したデータ挿入とスキーマレスのデータ挿入に対応。
• ClickHouse のすべてのデータ型をサポートします。

HTTP(S) で ClickHouse に接続するには、次の情報が必要です。

ClickHouse Cloud サービスの詳細は、ClickHouse Cloud コンソールで確認できます。 サービスを選択し、Connect をクリックします。

HTTPS を選択します。接続情報は curl コマンドの例として表示されます。

セルフマネージド ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。

このコネクタは、プラグインの実行に必要なすべてのクラスファイルを含む単一の JAR ファイルとして配布されます。

プラグインをインストールするには、次の手順に従ってください。

• ClickHouse Kafka Connect Sink リポジトリの Releases ページから、コネクタの JAR ファイルを含む ZIP アーカイブをダウンロードします。
• ZIP ファイルの内容を展開し、任意の場所にコピーします。
• Confluent Platform がプラグインを検出できるように、Connect のプロパティファイル内の plugin.path 設定に、プラグインディレクトリへのパスを追加します。
• 設定で topic 名、ClickHouse インスタンスの hostname、パスワードを指定します。

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
jdbcConnectionProperties=?sslmode=STRICT
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

• Confluent Platform を再起動します。
• Confluent Platform を使用している場合は、Confluent Control Center UI にログインし、利用可能なコネクタの一覧に ClickHouse Sink が含まれていることを確認します。

ClickHouse Sink を ClickHouse サーバー に接続するには、次の情報を指定する必要があります。

• 接続情報: ホスト名 (必須) とポート (任意)
• ユーザー認証情報: パスワード (必須) とユーザー名 (任意)
• コネクタクラス: com.clickhouse.kafka.connect.ClickHouseSinkConnector (必須)
• topics または topics.regex: ポーリングする Kafka トピック。トピック名はテーブル名と一致している必要があります (必須)
• キーコンバータと値コンバータ: トピック上のデータの種類に応じて設定します。worker の設定ですでに定義されていない場合は必須です。

設定オプションの完全な一覧:

ClickHouse Connect Sink は Kafka のトピックからメッセージを読み取り、適切なテーブルに書き込みます。ClickHouse Connect Sink は既存のテーブルにデータを書き込むため、データの insert を開始する前に、適切なスキーマを持つターゲットテーブルが ClickHouse に作成されていることを確認してください。

各トピックには、ClickHouse 内に専用のターゲットテーブルが必要です。ターゲットテーブル名はソースのトピック名と一致している必要があります。

ClickHouse Kafka Connect Sink に送信する前にメッセージを変換する必要がある場合は、Kafka Connect Transformations を使用してください。

スキーマが定義されている場合:

• (1) - JSON がサポートされるのは、ClickHouse の設定で input_format_binary_read_json_as_string=1 が指定されている場合のみです。これは RowBinary フォーマットファミリーでのみ機能し、この設定は insert リクエスト内のすべてのカラムに影響するため、それらはすべて文字列である必要があります。この場合、コネクタは STRUCT を JSON 文字列に変換します。

• (2) - struct に oneof のような ユニオン がある場合、コンバータはフィールド名に プレフィックス/接尾辞 を追加しないよう設定する必要があります。generate.index.for.unions=false の ProtobufConverter 用設定 があります。

スキーマが定義されていない場合:

レコードは JSON に変換され、JSONEachRow フォーマットの値として ClickHouse に送信されます。

すぐに使い始められるよう、よく使われる設定例をいくつか紹介します。

まず使い始めるための最も基本的な構成です。Kafka Connect を分散モードで実行し、localhost:8443 で SSL を有効にした ClickHouse サーバーが稼働しており、データがスキーマレスの JSON 形式であることを前提としています。

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "consumer.override.max.poll.records": "5000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "database": "default",
    "errors.retry.timeout": "60000",
    "exactlyOnce": "false",
    "hostname": "localhost",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "<PASSWORD>",
    "topics": "<TOPIC_NAME>",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
  }
}

コネクタは複数のトピックからデータを取り込むことができます

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "topics": "SAMPLE_TOPIC, ANOTHER_TOPIC, YET_ANOTHER_TOPIC",
    ...
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "<DLQ_TOPIC>",
    "errors.deadletterqueue.context.headers.enable": "true",
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

以下の型マッピングは、Kafka Connect における公式の Avro シリアライザー/デシリアライザー実装である io.confluent.connect.avro.AvroConverter で定義されています。変換ロジックの詳細については、Kafka Connect のドキュメントを参照してください。

✅: サポート

❌: 非サポート

️⚠️: 一部サポート

Kafka Connect の型と ClickHouse の型の対応については、サポートされているデータ型を参照してください。

次の Avro スキーマは、コネクタではサポートされていません。

• fixed の decimal 論理型

{"name": "decimal_18_4", "type": "fixed", "size": 8, "logicalType": "decimal", "precision": 18, "scale": 4}

• Nullable ユニオン

{"name": "mixed_union", "type": ["null", "string", "int"], "default": null}

• レコード型のユニオン

{
  "name": "record_union",
  "type": [
    {
      "type": "record",
      "name": "TypeA",
      "fields": [
        {
          "name": "label",
          "type": "string"
        }
      ]
    },
    {
      "type": "record",
      "name": "TypeB",
      "fields": [
        {
          "name": "count",
          "type": "int"
        }
      ]
    }
  ]
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

ご注意ください: クラスが見つからない問題が発生した場合、protobuf converter はすべての環境に含まれているわけではないため、依存関係をバンドルした別リリースの jar が必要になることがあります。

以下の型マッピングは、Kafka Connect の公式 Protobuf シリアライザー/デシリアライザー実装である io.confluent.connect.protobuf.ProtobufConverter で定義されています。変換ロジックの詳細については、Kafka Connect のドキュメントを参照してください。

✅: サポート対象

❌: サポート対象外

️⚠️: 一部サポート

Kafka Connect の型と ClickHouse の型マッピングについては、サポートされているデータ型を参照してください。

このコネクタは、Protobuf のユニオン (oneof) を ClickHouse の Variant 型に変換することをサポートしていません。代わりに、ClickHouse テーブルのスキーマでは、oneof フィールドを個別の Nullable フィールドとして列挙してください。

たとえば:

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message StringIntUnion {
  oneof mixed {
    string mixed_string = 2;
    int32 mixed_int = 3;
  }
}

次の ClickHouse テーブル定義に変換されます:

CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
    mixed_string Nullable(String),
    mixed_int Nullable(Int32)
) ENGINE = ...;

コネクタでは、以下の Protobuf スキーマはサポートされていません。

• 複数メッセージのユニオン (CH バージョン 26.1 より前)

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 を設定すると、このスキーマがサポートされます (こちらのドキュメントページを参照) 。

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  }
}

このコネクタは、ClickHouse のさまざまなフォーマットで String Converter をサポートしています：JSON、CSV、および TSV。

{
  "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 を指定してバッファリングを無効にしてください。

例:

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "exactlyOnce": "false",
    "bufferCount": "5000",
    "bufferFlushTime": "2000"
  }
}

ロギングは Kafka Connect Platform で自動的に提供されます。 ログの宛先とフォーマットは、Kafka Connect の設定ファイルで設定できます。

Confluent Platform を使用している場合は、CLI コマンドを実行してログを確認できます。

confluent local services connect log

詳細については、公式のチュートリアルを参照してください。

ClickHouse Kafka Connect は、Java Management Extensions (JMX) を通じてランタイムメトリクスを公開します。JMX は Kafka Connector でデフォルトで有効化されています。

このコネクタは、次のMBean名でカスタムメトリクスを公開します。

com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}

このコネクタは、データフロー、スループット、パフォーマンスの把握に役立つ、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 と統合されており、タスクのライフサイクルとエラー追跡に関するメトリクスを公開します。

タスクステータスのメトリクス:

• 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 設定ファイルを参照してください。

• 削除には対応していません。
• バッチサイズは 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 から 1 バッチごとに受け取るデータ量を制御することです。

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)

コネクタはフレームワークのバッファからメッセージをポーリングします。

• max.poll.records: 1 回のポーリングで返されるレコードの最大数 (デフォルト: 500)
• max.partition.fetch.bytes: パーティションごとの最大データ量 (デフォルト: 1048576 / 1 MB)

ClickHouseで最適なパフォーマンスを得るには、より大きなバッチを使用するようにしてください。

# ポーリングごとのレコード数を増やす
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

重要: Kafka Connect の fetch 設定は圧縮データを表しますが、ClickHouse が受信するのは非圧縮データです。これらの設定は、圧縮率を踏まえて調整してください。

トレードオフ:

• バッチが大きいほど = ClickHouse へのインジェスト性能が向上し、パーツ数が減り、オーバーヘッドが低減
• バッチが大きいほど = メモリ使用量が増え、エンドツーエンドのレイテンシが高くなる可能性
• バッチが大きすぎる場合 = タイムアウト、OutOfMemory エラー、または max.poll.interval.ms 超過のリスク

詳細: Confluent ドキュメント | Kafka ドキュメント

非同期挿入は、コネクタが比較的小さなバッチを送信する場合や、バッチ処理を ClickHouse に任せてインジェストをさらに最適化したい場合に有効な強力な機能です。

次のような場合は、非同期 INSERT を有効にすることを検討してください。

• 小さなバッチが多数ある: コネクタが小さなバッチ (1 バッチあたり 1000 行未満) を高頻度で送信している
• 高い並行性: 複数のコネクタ タスクが同じテーブルに書き込んでいる
• 分散デプロイ: 異なるホスト上で多数のコネクタ インスタンスを実行している
• パーツ作成のオーバーヘッド: 「パーツが多すぎる」エラーが発生している
• 混在ワークロード: リアルタイム インジェストとクエリ ワークロードを組み合わせている

次のような場合は、非同期 INSERT を使用しないでください。

• すでに大きなバッチ (1 バッチあたり 10,000 行超) を、頻度を制御しながら送信している
• データを即座に可視化する必要がある (クエリですぐにデータを参照できる必要がある)
• wait_for_async_insert=0 を使用する exactly-once セマンティクス が要件と競合する
• 代わりにクライアント側のバッチ処理の改善で効果を得られるユースケースである

非同期挿入を有効にすると、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)

これにより、作成されるパーツ数が大幅に減り、全体のスループットが向上します。

clickhouseSettings 設定パラメータに、非同期 INSERT の設定を追加します。

{
  "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 のフラッシュ動作は細かく調整できます：

"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 の場合、サーバークラッシュ時にデータが失われる可能性があること、大きなバッファによってメモリが逼迫する可能性があること

非同期 INSERT で exactlyOnce=true を使用する場合:

{
  "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 を参照してください。

スループットを向上させるには、並列度を上げます。

"tasks.max": "4"

各タスクは、トピックパーティションの一部を処理します。タスク数を増やすほど並列性は高まりますが、次の点に注意してください。

• 実質的に有効なタスク数の上限 = トピックパーティション数
• 各タスクは ClickHouse への接続をそれぞれ維持します
• タスク数が増えるほどオーバーヘッドが大きくなり、リソース競合が発生する可能性があります

推奨: まず tasks.max をトピックパーティション数と同じ値に設定し、その後 CPU 使用率とスループットのメトリクスを見ながら調整してください。

デフォルトでは、コネクタはパーティションごとにメッセージをバッチ処理します。スループットを高めるには、パーティションをまたいでバッチ処理できます。

"ignorePartitionsWhenBatching": "true"

** 警告**: exactlyOnce=false の場合にのみ使用してください。この設定では、より大きなバッチを作成してスループットを向上できますが、パーティションごとの順序保証は失われます。

コネクタが複数のトピックを購読するように設定されており、topic2TableMap を使ってトピックをテーブルにマッピングしていて、挿入時のボトルネックによってコンシューマラグが発生している場合は、代わりにトピックごとにコネクタを 1 つずつ作成することを検討してください。

これが発生する主な理由は、現時点ではバッチが各テーブルに直列で挿入されるためです。

推奨事項: 高ボリュームのトピックが複数ある場合は、並列挿入スループットを最大化するために、トピックごとにコネクタインスタンスを 1 つずつデプロイしてください。

ユースケースに適した ClickHouse テーブルエンジンを選択してください。

• MergeTree: ほとんどのユースケースに最適で、クエリと insert のパフォーマンスのバランスに優れています
• ReplicatedMergeTree: 高可用性に必要ですが、レプリケーションのオーバーヘッドが増えます
• 適切な ORDER BY を設定した *MergeTree: クエリパターンに合わせて最適化できます

検討すべき設定:

CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS 
    -- 並列パート書き込みのために挿入スレッドの最大数を増やす
    max_insert_threads = 4,
    -- 信頼性向上のためにクォーラムを使用した挿入を許可する（ReplicatedMergeTree）
    insert_quorum = 2

コネクタレベルの insert 設定:

"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"

コネクタは ClickHouse への HTTP 接続を維持します。遅延の大きいネットワークでは、タイムアウトを調整してください。

"jdbcConnectionProperties": "?socket_timeout=300000,connection_timeout=30000"

• socket_timeout (デフォルト: 30000 ms) : 読み取り操作の最大待機時間
• connection_timeout (デフォルト: 10000 ms) : 接続確立までの最大待機時間

大きなバッチでタイムアウトエラーが発生する場合は、これらの値を増やしてください。

クライアント V2 ("client_version": "V2") を使用する場合、データは client_network_buffer_size (バイト単位) で制御されるバッファを介して、ソケットとアプリケーションメモリの間でコピーされます。デフォルトのサイズは 300000 バイトです。バッファを大きくすると、高レイテンシまたは高帯域幅の接続でスループットが向上する可能性がありますが、接続タスクあたりのメモリ使用量が増加します。

すべてのクライアントオプションは、clickhouseSettings ではなく jdbcConnectionProperties で設定します。例:

"jdbcConnectionProperties": "?client_network_buffer_size=1000000"

複数のクライアントプロパティを設定する場合は、& で連結します。

"jdbcConnectionProperties": "?ssl=true&client_network_buffer_size=1000000"

以下の主要なメトリクスを監視します。

1. コンシューマラグ: Kafka の監視ツールを使用して、パーティションごとのラグを追跡します
2. コネクタのメトリクス: JMX 経由で receivedRecords、recordProcessingTime、taskProcessingTime を監視します (監視を参照)
3. ClickHouse のメトリクス:
   • system.asynchronous_inserts: 非同期 INSERT バッファの使用状況を監視します
   • system.parts: マージの問題を検出するためにパーツ数を監視します
   • system.merges: 実行中のマージを監視します
   • system.events: InsertedRows、InsertedBytes、FailedInsertQuery を追跡します

一般的なパフォーマンスの問題:

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. 十分にテストする: 本番環境にデプロイする前に、現実的な負荷をかけて設定変更を必ずテストする

以下に、高スループット向けに最適化した完全な例を示します。

{
  "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": "<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"
  }
}

この設定:

• 1 回の poll で最大 10,000 件のレコードを処理します
• より大きな INSERT を行うため、複数のパーティションにまたがってバッチ化します
• 16 MB のバッファで非同期 INSERT を使用します
• 8 個の task を並列実行します (パーティション数に合わせてください)
• 厳密な順序性よりもスループットを重視して最適化されています

これは、KeeperMap に保存されているオフセットと Kafka に保存されているオフセットが一致しない場合に発生します。通常は、トピックが削除されたか オフセットが手動で調整された場合です。 これを修正するには、該当するトピックとパーティションに保存されている古い値を削除する必要があります:

-- まず、データの保存に使用されているデータベースを特定します。
SELECT * FROM [database].connect_state

-- トピックとパーティションに一致するキーを特定します。
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]

現在は、一時的で再試行可能なエラーの特定に重点を置いており、以下が含まれます。

• 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 を追加することです。

transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_

これにより、データはネストされた JSON からフラット化された JSON に変換されます (区切り文字として _ を使用) 。その結果、テーブル内のフィールドは "field1_field2_field3" という形式 (つまり "before_id"、"after_id" など) に従います。

Kafka のキーはデフォルトでは value フィールドに保存されませんが、KeyToValue 変換を使うと、キーを value フィールド内の新しい _key フィールドに移動できます。

transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key