ClickHouse Kafka Connect Sink - ClickHouse Documentation

サポートが必要な場合は、リポジトリで issue を報告するか、ClickHouse の公開 Slack で質問してください。

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

ライセンス

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

環境要件

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

バージョン互換性マトリックス

ClickHouse Kafka Connect バージョン	ClickHouse バージョン	Kafka Connect	Confluent Platform
1.0.0	> 23.3	> 2.7	> 6.1

主な機能

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

インストール手順

接続情報を確認する

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

Parameter(s)	Description
`HOST` and `PORT`	通常、TLS を使用する場合のポートは 8443、TLS を使用しない場合は 8123 です。
`DATABASE NAME`	デフォルトでは `default` という名前のデータベースがあります。接続先のデータベース名を使用してください。
`USERNAME` and `PASSWORD`	デフォルトのユーザー名は `default` です。用途に応じたユーザー名を使用してください。

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 の設定ですでに定義されていない場合は必須です。

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

プロパティ名	説明	デフォルト値
`hostname` (必須)	サーバーのホスト名または IP アドレス	該当なし
`port`	ClickHouse のポート。デフォルトは 8443 (Cloud での HTTPS 用) ですが、HTTP (セルフホストでのデフォルト) の場合は 8123 にする必要があります	`8443`
`ssl`	ClickHouse への SSL 接続を有効にします	`true`
`jdbcConnectionProperties`	ClickHouse に接続する際の接続プロパティ。`?` で始め、各 `param=value` は `&` で連結する必要があります	`""`
`username`	ClickHouse データベースのユーザー名	`default`
`password` (必須)	ClickHouse データベースのパスワード	N/A
`database`	ClickHouse データベース名	`default`
`connector.class` (必須)	コネクタクラス (明示的に設定し、デフォルト値のまま使用)	`"com.clickhouse.kafka.connect.ClickHouseSinkConnector"`
`tasks.max`	コネクタのタスク数	`"1"`
`errors.retry.timeout`	Kafka Connect の最大再試行継続時間 (ミリ秒単位) 。`0` は再試行なし。`-1` は無限に再試行。推奨値は “10000” ms (10 秒) 超ですタイムアウト	`"0"`
`exactlyOnce`	Exactly Once の有効化	`"false"`
`topics` (必須)	ポーリングする Kafka トピック - トピック名はテーブル名と一致している必要があります	`""`
`key.converter` (必須* - 説明を参照)	キーの型に応じて設定します。キーを渡す場合 (worker config で定義されていない場合) は、ここでの設定が必須です。	`"org.apache.kafka.connect.storage.StringConverter"`
`value.converter` (必須* - 説明を参照)	トピック上のデータの型に応じて設定します。対応フォーマット: JSON、String、Avro、Protobuf。worker config で定義されていない場合は、ここでの設定が必須です。	`"org.apache.kafka.connect.json.JsonConverter"`
`value.converter.schemas.enable`	コネクタの値コンバータにおけるスキーマサポート	`"false"`
`errors.tolerance`	コネクタのエラー許容設定。対応: none、all	`"none"`
`errors.deadletterqueue.topic.name`	設定すると (errors.tolerance=all の場合) 、失敗したバッチは DLQ に送られます (トラブルシューティングを参照)	`""`
`errors.deadletterqueue.context.headers.enable`	DLQ に追加のヘッダーを追加します	`""`
`clickhouseSettings`	ClickHouse 設定のカンマ区切りリスト (例: “insert_quorum=2, etc…”)	`""`
`topic2TableMap`	トピック名をテーブル名に対応付けるカンマ区切りリスト (例: “topic1=table1, topic2=table2, etc…”)	`""`
`tableRefreshInterval`	テーブル定義 cache を更新する間隔 (秒)	`0`
`keeperOnCluster`	セルフホストのインスタンスで、exactly-once の connect_state テーブルに対する `ON CLUSTER` パラメーター (例: `ON CLUSTER clusterNameInConfigFileDefinition`) を設定できます (Distributed DDL Queries を参照	`""`
`bypassRowBinary`	スキーマベースのデータ (Avro、Protobuf など) に対する RowBinary および RowBinaryWithDefaults の使用を無効にできます。データに欠落したカラムがあり、Nullable/Default を許容できない場合にのみ使用してください	`"false"`
`dateTimeFormats`	DateTime64 スキーマのフィールドをパースするための日時フォーマット。各フォーマットは `;` で区切ります (例: `someDateField=yyyy-MM-dd HH:mm:ss.SSSSSSSSS;someOtherDateField=yyyy-MM-dd HH:mm:ss`) 。	`""`
`tolerateStateMismatch`	コネクタが、AFTER_PROCESSING に保存されている現在のオフセットよりも”前”のレコードを破棄できるようにします (たとえば、オフセット 5 が送信され、最後に記録されたオフセットが 250 だった場合) 。障害後のインジェストを復旧するために使用し、完了したら `"false"` に戻してください。	`"false"`
`ignorePartitionsWhenBatching`	insert 用にメッセージを収集する際、パーティションを無視します (ただし、`exactlyOnce` が `false` の場合のみ) 。パフォーマンスに関する注意: コネクタのタスク数が増えるほど、各タスクに割り当てられる Kafka パーティション数は少なくなるため、効果が頭打ちになる可能性があります。	`"false"`
`bufferCount` (v1.3.6以降)	ClickHouse にフラッシュする前に、メモリ内でバッファするレコード数です。`0` は内部バッファリングを無効にします。`exactlyOnce=true` ではバッファリングはサポートされていません。	`"0"`
`bufferFlushTime` (v1.3.6以降)	`exactlyOnce=false` の場合に、flush 前にレコードをバッファに保持する最大時間 (ミリ秒) です。`0` を指定すると、時間ベースのフラッシュは無効になります。デフォルト値は `0` です。時間ベースのしきい値を使用する場合にのみ必要です。`bufferCount > 0` の場合にのみ有効です。	`"0"`
`reportInsertedOffsets` (v1.3.6以降)	`exactlyOnce=false` の場合、`preCommit` は `currentOffsets` ではなく、正常に挿入されたオフセットのみを返すようになります。これは `ignorePartitionsWhenBatching=true` の場合には適用されず、その場合は引き続き `currentOffsets` が返されます。	`"false"`

ターゲットテーブル

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

前処理

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

サポートされているデータ型

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

Kafka Connect Type	ClickHouse 型	サポート	Primitive
STRING	String	✅	はい
STRING	JSON. 以下の (1) を参照	✅	はい
INT8	Int8	✅	はい
INT16	Int16	✅	はい
INT32	Int32	✅	はい
INT64	Int64	✅	はい
FLOAT32	Float32	✅	はい
FLOAT64	Float64	✅	はい
BOOLEAN	Boolean	✅	はい
ARRAY	Array(T)	✅	いいえ
MAP	Map(Primitive, T)	✅	いいえ
STRUCT	Variant(T1, T2, …)	✅	いいえ
STRUCT	Tuple(a T1, b T2, …)	✅	いいえ
STRUCT	Nested(a T1, b T2, …)	✅	いいえ
STRUCT	JSON. 以下の (1), (2) を参照	✅	いいえ
BYTES	String	✅	いいえ
org.apache.kafka.connect.data.Time	Int64 / DateTime64	✅	いいえ
org.apache.kafka.connect.data.Timestamp	Int32 / Date32	✅	いいえ
org.apache.kafka.connect.data.Decimal	Decimal	✅	いいえ

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

上記のコネクタ設定では、connector.client.config.override.policy=All を使用して、ワーカー設定でクライアントのオーバーライドを有効にする必要があります。詳細については、Kafka Connect のドキュメントを参照してください。

複数のトピックを使用する基本構成

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

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

DLQ を使用する基本設定

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

さまざまなデータフォーマットでの使用

Avro スキーマの対応

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

Avro 型マッピング

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

Avro Type	Kafka Connect Type	Supported	Notes
null	N/A	❌	単独の型としてはサポートされていませんが、ユニオン内では使用できます
boolean	BOOLEAN	✅
int	INT8/INT16/INT32	✅	デフォルトは INT32 です。スキーマにプロパティ `connect.type=int8` がある場合は INT8 として扱われます (`connect.type=int16` の場合は同様に INT16)
long	INT64	✅
float	FLOAT32	✅
double	FLOAT64	✅
bytes	BYTES	✅
string	STRING	✅
record	STRUCT	✅
enum	STRING	✅
array	ARRAY/MAP	✅	デフォルトは ARRAY です。フィールドがもともと `AvroData.fromConnectSchema` によって構築されていた場合は、MAP として扱われます (source)
map	MAP	✅
union	STRUCT/`<T>`	⚠️	デフォルトは STRUCT です。`flatten.singleton.unions=true` の場合は、ユニオン定義内の単一型 `T` として扱われます (docsを参照)
fixed	BYTES	⚠️	fixed `decimal` 論理型はサポートされていません (詳細は以下を参照)

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

サポートされていない Avro スキーマ

次の 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"
        }
      ]
    }
  ]
}

Protobuf スキーマ対応

{
  "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 が必要になることがあります。

Protobuf 型マッピング

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

Protobuf Type	Kafka Connect Type	Supported	Notes
double	FLOAT64	✅
float	FLOAT32	✅
int32	INT8/INT16/INT32	✅	デフォルトは INT32 です。スキーマにオプション `connect.type=int8` がある場合は INT8 になります (`connect.type=int16` の場合も同様に INT16)
sint32	INT8/INT16/INT32	✅	デフォルトは INT32 です。スキーマにオプション `connect.type=int8` がある場合は INT8 になります (`connect.type=int16` の場合も同様に INT16)
sfixed32	INT8/INT16/INT32	✅	デフォルトは INT32 です。スキーマにオプション `connect.type=int8` がある場合は INT8 になります (`connect.type=int16` の場合も同様に INT16)
uint32	INT64	✅
fixed32	INT64	✅
int64	INT64	✅
uint64	INT64	✅
sint64	INT64	✅
fixed64	INT64	✅
sfixed64	INT64	✅
bool	BOOLEAN	✅
string	STRING	✅
bytes	BYTES	✅
enum	INT32/STRING	✅	デフォルトは STRING です。`int.for.enums=true` の場合は INT32 になります (スキーマレジストリのドキュメントを参照)
message	STRUCT	⚠️	以下の未サポートのスキーマのセクションを参照
repeated T (where T is not a map entry)	ARRAY	✅
`map<K, V>`	MAP	✅
oneof	STRUCT	⚠️	oneof を ClickHouse スキーマに変換する方法については、以下のセクションを参照
google.protobuf.DoubleValue	FLOAT64	✅
google.protobuf.FloatValue	FLOAT32	✅
google.protobuf.Int64Value	INT64	✅
google.protobuf.UInt64Value	INT64	✅
google.protobuf.UInt32Value	INT64	✅
google.protobuf.Int32Value	INT32	✅
google.protobuf.BoolValue	BOOLEAN	✅
google.protobuf.StringValue	STRING	✅
google.protobuf.BytesValue	BYTES	✅
google.protobuf.Timestamp	org.apache.kafka.connect.data.Timestamp	✅
google.type.Date	org.apache.kafka.connect.data.Date	✅
google.type.TimeOfDay	org.apache.kafka.connect.data.Time	✅

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

`oneof` フィールドを 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 スキーマ

コネクタでは、以下の 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;
  }

  // ClickHouse の Nullable(Tuple(label String)) に変換されますが、サポートされていません
  message TypeA {
    string label = 1;
  }

  // ClickHouse の Nullable(Tuple(count Int32)) に変換されますが、サポートされていません
  message TypeB {
    int32 count = 1;
  }
}

CH バージョン 26.1 以降では、allow_experimental_nullable_tuple_type=1 を設定すると、このスキーマがサポートされます (こちらのドキュメントページを参照) 。

JSON スキーマのサポート

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

Stringのサポート

このコネクタは、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 でデフォルトで有効化されています。

ClickHouse固有のメトリクス

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

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: ポーリングされたレコード総数

監視のベストプラクティス

コンシューマラグを監視する: 処理のボトルネックを特定するため、パーティションごとの records-lag を追跡します
エラー率を追跡する: データ品質の問題を検出するため、errors-total と records-skip-total を監視します
タスクの健全性を確認する: タスクが正常に実行されていることを確認するため、タスクのステータスメトリクスを監視します
スループットを測定する: インジェストのパフォーマンスを追跡するため、records-send-rate と byte-rate を使用します
接続の健全性を監視する: ネットワークの問題を確認するため、ノードレベルの接続メトリクスを確認します
圧縮効率を追跡する: データ転送を最適化するため、compression-rate を使用します

JMX メトリクスの詳細な定義と Prometheus インテグレーションについては、jmx-export-connector.yml 設定ファイルを参照してください。

制限事項

削除には対応していません。
バッチサイズは Kafka Consumer のプロパティを継承します。
exactly-once のために KeeperMap を使用している場合、オフセットを変更または巻き戻したときは、そのトピックに対応する KeeperMap の内容を削除する必要があります。 (詳細は下記のトラブルシューティングガイドを参照してください)

パフォーマンスチューニングとスループット最適化

このセクションでは、ClickHouse Kafka Connect Sink のパフォーマンスチューニングの方法について説明します。パフォーマンスチューニングは、高スループットのユースケースに対応する場合や、リソース使用率を最適化して遅延を最小限に抑える必要がある場合に不可欠です。

パフォーマンスチューニングが必要になるのはどのような場合ですか？

パフォーマンスチューニングは、通常、次のような状況で必要になります。

高スループットのワークロード: Kafkaトピックから毎秒数百万件のイベントを処理する場合
コンシューマラグ: データ生成の速度にコネクタが追いつかず、ラグが増大している場合
リソース制約: CPU、メモリ、またはネットワークの使用を最適化する必要がある場合
複数のトピック: 大量のデータが流れる複数のトピックを同時に消費する場合
小さなメッセージサイズ: サーバー側のバッチ処理の恩恵を受けられる、小さなメッセージを大量に扱う場合

パフォーマンスチューニングが通常は不要なのは、次のような場合です。

低〜中程度の量 (毎秒10,000メッセージ未満) を処理している
コンシューマラグが安定しており、ユースケース上許容できる範囲に収まっている
デフォルトのコネクタ設定ですでに必要なスループット要件を満たしている
ClickHouseクラスターが流入する負荷を容易に処理できる

データフローを理解する

チューニングを行う前に、データがコネクタ内をどのように流れるかを理解しておくことが重要です。

Kafka Connect Framework がバックグラウンドで Kafka トピックからメッセージを取得します
コネクタ がフレームワークの内部バッファからメッセージをポーリングします
コネクタ はポーリングサイズに基づいてメッセージをバッチ化します
ClickHouse は HTTP/S 経由でバッチ化された insert を受け取ります
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で最適なパフォーマンスを得るには、より大きなバッチを使用するようにしてください。

# ポーリングごとのレコード数を増やす
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 ドキュメントを参照してください。

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

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

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

非同期挿入

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

非同期 INSERT を使用するタイミング

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

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

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

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

非同期 INSERT の仕組み

非同期挿入を有効にすると、ClickHouse は次のように動作します。

コネクタから INSERT クエリを受け取ります
データをメモリ上のバッファに書き込みます (すぐにディスクへは書き込みません)
コネクタに成功を返します (wait_for_async_insert=0 の場合)
次のいずれかの条件が満たされると、バッファをディスクにフラッシュします。
- バッファが async_insert_max_data_size に達する (デフォルト: 100 MB)
- 最初の INSERT から async_insert_busy_timeout_ms ミリ秒が経過する (デフォルト: 1000 ms)
- 累積されたクエリ数が上限に達する (async_insert_max_query_number、デフォルト: 100)

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

非同期 INSERT を有効にする

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 の動作の調整

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

exactly-once セマンティクスの非同期 INSERT

非同期 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 テーブルエンジンに関する考慮事項

ユースケースに適した 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 接続を維持します。遅延の大きいネットワークでは、タイムアウトを調整してください。

"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"

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

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

パフォーマンスの監視とトラブルシューティング

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

コンシューマラグ: Kafka の監視ツールを使用して、パーティションごとのラグを追跡します
コネクタのメトリクス: JMX 経由で receivedRecords、recordProcessingTime、taskProcessingTime を監視します (監視を参照)
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 回の insert あたり 10,000～100,000 行を目安にする
非同期 INSERT を使用する: 小さなバッチを多数送信する場合や、高い同時実行性が求められる場合に使用する
exactly-once セマンティクスでは常に wait_for_async_insert=1 を使用する
水平スケールする: tasks.max をパーティション数まで増やす
スループットの高いトピックごとに 1 つのコネクタを使用する: スループットを最大化するため
継続的に監視する: コンシューマラグ、パーツ数、merge アクティビティを追跡する
十分にテストする: 本番環境にデプロイする前に、現実的な負荷をかけて設定変更を必ずテストする

例: 高スループット構成

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

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

上記のコネクタ設定では、worker の設定で connector.client.config.override.policy=All を指定し、クライアント設定のオーバーライドを有効にする必要があります。詳しくは、Kafka Connect のドキュメントを参照してください。

この設定:

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

トラブルシューティング

”トピック `[someTopic]` パーティション `[0]` の状態不一致”

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

-- まず、データの保存に使用されているデータベースを特定します。
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 を追加することです。

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 フィールドに移動できます。

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

最終更新日 2026年6月25日

Confluent Cloud 上の Kafka Connector SinkConfluent Cloud で完全マネージド型の ClickHouse Connector Sink を使用するためのガイド

​ライセンス

​環境要件

​バージョン互換性マトリックス

​主な機能

​インストール手順

​接続情報を確認する

​一般的なインストール手順

​設定オプション

​ターゲットテーブル

​前処理

​サポートされているデータ型

​設定レシピ

​基本構成

​複数のトピックを使用する基本構成

​DLQ を使用する基本設定

​さまざまなデータフォーマットでの使用

Avro スキーマの対応

Avro 型マッピング

サポートされていない Avro スキーマ

Protobuf スキーマ対応

Protobuf 型マッピング

oneof フィールドを ClickHouse のカラムに変換する際の注意

サポートされていない Protobuf スキーマ

JSON スキーマのサポート

Stringのサポート

​内部バッファリング

​ロギング

​監視

​ClickHouse固有のメトリクス

​Kafka Producer/Consumer メトリクス

​Kafka Connect Framework のメトリクス

​監視のベストプラクティス

​制限事項

​パフォーマンスチューニングとスループット最適化

​パフォーマンスチューニングが必要になるのはどのような場合ですか？

​データフローを理解する

​Kafka Connect のバッチサイズ調整

fetch設定

ポーリング設定

高スループット向けの推奨設定

​非同期挿入

非同期 INSERT を使用するタイミング

非同期 INSERT の仕組み

非同期 INSERT を有効にする

非同期 INSERT の動作の調整

exactly-once セマンティクスの非同期 INSERT

​コネクタの並列度

コネクタあたりのタスク数

バッチ処理時にパーティションを区別しない

​複数の高スループットなトピック

​ClickHouse テーブルエンジンに関する考慮事項

​接続プーリングとタイムアウト

​パフォーマンスの監視とトラブルシューティング

​ベストプラクティスの要約

​例: 高スループット構成

​トラブルシューティング

​”トピック [someTopic] パーティション [0] の状態不一致”

​“コネクタはどのようなエラーで再試行しますか？”

​“すべてのデータが空白/ゼロになっている”

​“ClickHouse で Kafka のキーを使いたい”

ライセンス

環境要件

バージョン互換性マトリックス

主な機能

インストール手順

接続情報を確認する

一般的なインストール手順

設定オプション

ターゲットテーブル

前処理

サポートされているデータ型

設定レシピ

基本構成

複数のトピックを使用する基本構成

DLQ を使用する基本設定

さまざまなデータフォーマットでの使用

`oneof` フィールドを ClickHouse のカラムに変換する際の注意

内部バッファリング

ロギング

監視

ClickHouse固有のメトリクス

Kafka Producer/Consumer メトリクス

Kafka Connect Framework のメトリクス

監視のベストプラクティス

制限事項

パフォーマンスチューニングとスループット最適化

パフォーマンスチューニングが必要になるのはどのような場合ですか？

データフローを理解する

Kafka Connect のバッチサイズ調整

非同期挿入

コネクタの並列度

複数の高スループットなトピック

ClickHouse テーブルエンジンに関する考慮事項

接続プーリングとタイムアウト

パフォーマンスの監視とトラブルシューティング

ベストプラクティスの要約

例: 高スループット構成

トラブルシューティング

”トピック `[someTopic]` パーティション `[0]` の状態不一致”

“コネクタはどのようなエラーで再試行しますか？”

“すべてのデータが空白/ゼロになっている”

“ClickHouse で Kafka のキーを使いたい”