AvroConfluent | ClickHouse Docs

入力	出力	エイリアス
✔	✔

説明

Apache Avro は、効率的なデータ処理のためにバイナリエンコーディングを使用する、行指向のシリアライゼーションフォーマットです。AvroConfluent フォーマットは、Confluent スキーマレジストリ (または API 互換のサービス) を使用した、Avro エンコード済みメッセージの読み書きをサポートします。

各メッセージは Confluent のワイヤ形式に従います。これは、マジックバイト (0x00) に続いて 4 バイトのビッグエンディアンのスキーマ ID が配置され、その後に Avro のバイナリデータが続く形式です。読み取り時には、ClickHouse はレジストリに問い合わせてスキーマ ID を解決します。書き込み時には、ClickHouse は出力カラムから導出したスキーマを登録し、生成された ID を各行の先頭に付加します。最適なパフォーマンスを得るため、スキーマはキャッシュされます。

データ型マッピング

次の表は、Apache Avro フォーマットでサポートされるすべてのデータ型と、INSERT および SELECT クエリでそれぞれに対応する ClickHouse のデータ型を示しています。

Avro data type `INSERT`	ClickHouse data type	Avro data type `SELECT`
`boolean`, `int`, `long`, `float`, `double`	Int(8\16\32), UInt(8\16\32)	`int`
`boolean`, `int`, `long`, `float`, `double`	Int64, UInt64	`long`
`boolean`, `int`, `long`, `float`, `double`	Float32	`float`
`boolean`, `int`, `long`, `float`, `double`	Float64	`double`
`bytes`, `string`, `fixed`, `enum`	String	`bytes` または `string` *
`bytes`, `string`, `fixed`	FixedString(N)	`fixed(N)`
`enum`	Enum(8\16)	`enum`
`array(T)`	Array(T)	`array(T)`
`map(V, K)`	Map(V, K)	`map(string, K)`
`union(null, T)`, `union(T, null)`	Nullable(T)	`union(null, T)`
`union(T1, T2, …)` **	Variant(T1, T2, …)	`union(T1, T2, …)` **
`null`	Nullable(Nothing)	`null`
`int (date)` ***	Date, Date32	`int (date)` ***
`long (timestamp-millis)` ***	DateTime64(3)	`long (timestamp-millis)` ***
`long (timestamp-micros)` ***	DateTime64(6)	`long (timestamp-micros)` ***
`bytes (decimal)` ***	DateTime64(N)	`bytes (decimal)` ***
`int`	IPv4	`int`
`fixed(16)`	IPv6	`fixed(16)`
`bytes (decimal)` ***	Decimal(P, S)	`bytes (decimal)` ***
`string (uuid)` ***	UUID	`string (uuid)` ***
`fixed(16)`	Int128/UInt128	`fixed(16)`
`fixed(32)`	Int256/UInt256	`fixed(32)`
`record`	Tuple	`record`

bytes がデフォルトで、設定 output_format_avro_string_column_pattern で制御されます

** Variant 型は null をフィールド値として暗黙的に受け入れるため、たとえば Avro の union(T1, T2, null) は Variant(T1, T2) に変換されます。そのため、ClickHouse から Avro を生成する際には、Avro の union 型の候補に常に null 型を含める必要があります。これは、スキーマ推論の時点では、実際に null となる値があるかどうか分からないためです。

*** Avro 論理型

サポートされていない Avro 論理データ型:

time-millis
time-micros
duration

フォーマット設定

Setting	Description	Default
`input_format_avro_allow_missing_fields`	フィールドがスキーマ内に見つからない場合に、エラーを発生させる代わりにデフォルト値を使用するかどうか。	`0`
`input_format_avro_null_as_default`	`null` 値を非 Nullable のカラムに挿入する際に、エラーを発生させる代わりにデフォルト値を使用するかどうか。	`0`
`format_avro_schema_registry_url`	Confluent スキーマレジストリの URL。基本認証では、URL エンコードされた認証情報を URL パスに直接含めることができます。
`format_avro_schema_registry_connection_timeout`	スキーマレジストリ HTTP クライアントの接続タイムアウト (スキーマ取得と登録の両方で使用) の秒数。0 より大きく、600 (10 分) 未満である必要があります。	`1`
`format_avro_schema_registry_send_timeout`	スキーマレジストリ HTTP クライアントの送信タイムアウトの秒数。0 より大きく、600 (10 分) 未満である必要があります。	`1`
`format_avro_schema_registry_receive_timeout`	スキーマレジストリ HTTP クライアントの受信タイムアウトの秒数。0 より大きく、600 (10 分) 未満である必要があります。	`1`
`output_format_avro_confluent_subject`	出力用: スキーマレジストリでスキーマを登録する subject 名。書き込み時に必須です。
`output_format_avro_string_column_pattern`	出力用: Avro `string` としてシリアライズする String カラムの正規表現 (デフォルトは `bytes`) 。

例

Kafka からの読み込み

Kafka table engine を使用して Avro でエンコードされた Kafka トピックを読み込むには、format_avro_schema_registry_url 設定でスキーマレジストリの URL を指定します。

CREATE TABLE topic1_stream
(
    field1 String,
    field2 String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-broker',
kafka_topic_list = 'topic1',
kafka_group_name = 'group1',
kafka_format = 'AvroConfluent',
format_avro_schema_registry_url = 'http://schema-registry-url';

SELECT * FROM topic1_stream;

Kafka への書き込み

AvroConfluent メッセージを Kafka トピックに書き込むには、スキーマレジストリの URL と subject 名の両方を設定します。スキーマは最初の書き込み時に、自動的にレジストリへ登録されます。

CREATE TABLE topic1_sink
(
    field1 String,
    field2 String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-broker',
kafka_topic_list = 'topic1',
kafka_format = 'AvroConfluent',
format_avro_schema_registry_url = 'http://schema-registry-url',
output_format_avro_confluent_subject = 'topic1-value';

INSERT INTO topic1_sink VALUES ('hello', 'world');

Basic認証を使用する

スキーマレジストリでBasic認証が必要な場合 (例: Confluent Cloud を使用している場合) 、format_avro_schema_registry_url 設定に URL エンコードされた認証情報を指定できます。

CREATE TABLE topic1_stream
(
    field1 String,
    field2 String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-broker',
kafka_topic_list = 'topic1',
kafka_group_name = 'group1',
kafka_format = 'AvroConfluent',
format_avro_schema_registry_url = 'https://<username>:<password>@schema-registry-url';

トラブルシューティング

インジェストの進行状況を監視し、Kafkaコンシューマーのエラーをデバッグするには、system.kafka_consumers システムテーブルをクエリします。デプロイ環境に複数のレプリカがある場合 (例: ClickHouse Cloud) は、clusterAllReplicas テーブル関数を使用する必要があります。

SELECT * FROM clusterAllReplicas('default',system.kafka_consumers)
ORDER BY assignments.partition_id ASC;

スキーマ解決で問題が発生した場合は、kafkacat と clickhouse-local を使ってトラブルシューティングできます。

$ kafkacat -b kafka-broker  -C -t topic1 -o beginning -f '%s' -c 3 | clickhouse-local   --input-format AvroConfluent --format_avro_schema_registry_url 'http://schema-registry' -S "field1 Int64, field2 String"  -q 'select *  from table'
1 a
2 b
3 c