リファレンス

名前	ロゴ	Type	ステータス	説明
Apache Kafka		ストリーミング	安定版	ClickPipes を設定し、Apache Kafka から ClickHouse Cloud へのストリーミングデータの取り込みを開始します。
Confluent Cloud		ストリーミング	安定版	直接インテグレーションにより、Confluent と ClickHouse Cloud の連携を最大限に活用できます。
Redpanda		ストリーミング	安定版	ClickPipes を設定し、Redpanda から ClickHouse Cloud へのストリーミングデータの取り込みを開始します。
AWS MSK		ストリーミング	安定版	ClickPipes を設定し、AWS MSK から ClickHouse Cloud へのストリーミングデータの取り込みを開始します。
Azure Event Hubs		ストリーミング	安定版	ClickPipes を設定し、Azure Event Hubs から ClickHouse Cloud へのストリーミングデータの取り込みを開始します。
WarpStream		ストリーミング	安定版	ClickPipes を設定し、WarpStream から ClickHouse Cloud へのストリーミングデータの取り込みを開始します。

Kafkaコンシューマーの分離レベルは、ClickPipeがKafkaトランザクションでコミットされたメッセージのみを読み取り、挿入するかどうかを決定します。kafka_read_committed 設定を有効にすると、read_committed が使用され、中断されたKafkaトランザクションのメッセージはスキップされます。無効にすると、read_uncommitted が使用され、すべてのメッセージが読み取られます。この設定は、詳細設定 で構成できます。

対応しているフォーマットは次のとおりです。

• JSON
• AvroConfluent
• Protobuf

現在、ClickPipes では以下の標準的な ClickHouse データ型がサポートされています。

• 基本的な数値型 - [U]Int8/16/32/64、Float32/64、BFloat16
• 大きな整数型 - [U]Int128/256
• Decimal 型
• Boolean
• String
• FixedString
• Date、Date32
• DateTime、DateTime64 (UTC タイムゾーンのみ)
• Enum8/Enum16
• UUID
• IPv4
• IPv6
• Time、Time64
• JSON
• すべての ClickHouse LowCardinality 型
• キーと値に上記のいずれかの型を使用する Map (Nullable を含む)
• 要素に上記のいずれかの型を使用する Tuple と Array (Nullable を含む、深さは 1 レベルのみ)
• SimpleAggregateFunction 型 (AggregatingMergeTree または SummingMergeTree の宛先向け)

ClickPipes は、次の条件で Variant 型をサポートします。

• Avro のユニオン。Avro スキーマに複数の NULL 以外の型を含むユニオンがある場合、ClickPipes は適切な Variant 型を推論します。それ以外の Avro データでは、Variant 型はサポートされません。
• JSON フィールド。ソースデータストリーム内の任意の JSON フィールドに対して、Variant 型 (Variant(String, Int64, DateTime) など) を手動で指定できます。複雑なサブタイプ (Array/Map/Tuple) はサポートされません。また、ClickPipes では使用すべき Variant のサブタイプを判定する仕組み上、Variant の定義で使用できる整数型または datetime 型は 1 つのみです。たとえば、Variant(Int64, UInt32) はサポートされません。

ClickPipes は、次の状況で JSON 型をサポートします。

• Avro の Record フィールドと Protobuf の Message フィールドは、常に JSON カラムに割り当てることができます。
• Avro の String フィールドと Bytes フィールドは、Avro フィールドに実際には JSON オブジェクトを表す文字列が含まれている場合、JSON カラムに割り当てることができます。
• Protobuf の String 型と Bytes 型は、Protobuf フィールドに実際には JSON オブジェクトを表す文字列が含まれている場合、JSON カラムに割り当てることができます。
• 値が常に JSON オブジェクトである JSON フィールドは、宛先の JSON カラムに割り当てることができます。

fixed パスや skipped パスを含め、宛先カラムは目的の JSON 型に手動で変更する必要があることに注意してください。

ClickPipes は、local-timestamp-millis と local_timestamp-micros を除くすべての Avro Primitive 型、Complex 型、および Logical 型をサポートしています。Avro の record 型は Tuple に、array 型は Array に、map は Map に変換されます (キーは文字列のみ) 。一般的には、こちら に記載されている変換が利用可能です。ClickPipes は型変換時のオーバーフローや精度の損失をチェックしないため、Avro の数値型では厳密に対応する型を使用することを推奨します。また、すべての Avro 型は String カラムに挿入することもでき、その場合は有効な JSON 文字列として表されます。

Avro の Nullable 型は、T を基本の Avro 型として、(T, null) または (null, T) のユニオン スキーマを使って定義されます。スキーマ推論時には、このようなユニオンは ClickHouse の Nullable カラムにマッピングされます。なお、ClickHouse は Nullable(Array)、Nullable(Map)、Nullable(Tuple) 型をサポートしていません。これらの型に対する Avro の null ユニオンは、非 Nullable 型にマッピングされます (Avro の Record 型は ClickHouse の named Tuple にマッピングされます) 。これらの型における Avro の null は、次のように挿入されます。

• null の Avro array の場合は空の Array
• null の Avro Map の場合は空の Map
• null の Avro Record の場合は、すべてのフィールドがデフォルト値またはゼロ値の named Tuple

ClickPipes は、長らく非推奨となっている proto 2 の group 型を除き、Protobuf 2 および 3 のすべての型をサポートしています。基本的な型変換には、以下のマッピングが使用されます。

次の 既知の型 もサポートされています。

スキーマ推論では、Protobuf の oneof フィールドはデフォルトで名前付き Tuple にマッピングされ、非デフォルト値を持つフィールドは最大でも 1 つだけです。これらのフィールドは、設定されている構成フィールドの型をアクティブな値が取る Variant カラムに自動的にマッピングすることもできます。あるいは、各構成フィールドをそれぞれ個別の ClickHouse カラムに手動でマッピングすることもできます。oneof フィールドは相互排他的であるため、各レコードで値が入るカラムは 1 つだけです。

ClickPipe 用に定義された最上位の Protobuf スキーマに、それ自体が protobuf の Message である単一の repeated フィールドが含まれている場合、スキーマ推論とカラムマッピングは、その「内包された」Message フィールドに基づいて行われます。Kafka メッセージはそのようなメッセージのリストとして処理され、1 つの Kafka メッセージが複数の ClickHouse の行に展開されます。

以下の仮想カラムは、Kafka 互換のストリーミングデータソースでサポートされています。新しい宛先を作成する際は、Add Column ボタンを使用して仮想カラムをターゲットテーブルに追加できます。

_raw_message カラムは、JSON データに対してのみ推奨される点に注意してください。 JSON 文字列だけが必要なユースケースでは (たとえば、ClickHouse の JsonExtract* 関数を使用して 下流の materialized view にデータを投入する場合) 、「仮想ではない」カラムをすべて削除することで ClickPipes のパフォーマンスが向上することがあります。

_key を String カラムにマッピングすると、元の Kafka メッセージキーが保存されます。このマッピングでは、キーのデコードやスキーマのルックアップは行われません。構造化キーからフィールドを抽出するには、_key. で始まるソースフィールドをマッピングします。たとえば、JSON キー {"customer":{"id":42}} の場合、_key.customer.id を customer_id などの宛先カラムにマッピングします。

_key.* のマッピング、または String 以外の型への _key の直接マッピングでは、ClickPipes はまず、キーが設定済みのスキーマレジストリを使用してエンコードされているかどうかを確認します。エンコードされていない場合は、キーが JSON オブジェクトかどうかを確認します。レジストリでエンコードされたキーは、レコード値と同じフォーマットおよびレジストリファミリーを使用する必要がありますが、異なるスキーマ ID を使用できます。キースキーマの変更は自動的に検出されます。

デコードが有効な場合、生データのキー、JSON スカラーまたは配列、不正な JSON からはネストされたキーフィールドに値が設定されません。対応する宛先カラムには、型のデフォルト値が設定されます。ClickPipes がレジストリでエンコードされたキーを認識しても、そのスキーマを取得または適用できない場合は、Kafka レコードを ClickPipes エラーテーブルに書き込みます。

Avro キーでは、プリミティブ型、配列、map、fixed 型など、レコード以外のルートスキーマも使用できます。これをデコードするには、_key を String 以外の互換性のある宛先型に直接マッピングします。レコード以外のスキーマでは、ネストされたキーフィールドに値を設定できません。