AvroConfluent

入力	出力	エイリアス
✔	✗

説明

Apache Avro は、効率的なデータ処理のためにバイナリエンコードを使用する行指向のシリアル化フォーマットです。AvroConfluent フォーマットは、Confluent Schema Registry（またはその API 互換サービス）を用いてシリアル化された、単一オブジェクト形式の Avro でエンコードされた Kafka メッセージのデコードをサポートします。

各 Avro メッセージにはスキーマ ID が埋め込まれており、ClickHouse は設定済みの Schema Registry に対してクエリを実行することで自動的にスキーマを解決します。一度解決されたスキーマは、パフォーマンスを最適化するためにキャッシュされます。

データ型のマッピング

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

Avro データ型 INSERT	ClickHouse データ型	Avro データ型 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 type はフィールド値として暗黙的に null を受け入れるため、たとえば Avro の union(T1, T2, null) は Variant(T1, T2) に変換されます。 その結果、ClickHouse から Avro を生成する際には、スキーマ推論中に任意の値が実際に null かどうかを判断できないため、常に Avro の union 型の集合に null 型を含める必要があります。

*** Avro logical types

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

• time-millis
• time-micros
• duration

フォーマット設定

Setting	Description	Default
input_format_avro_allow_missing_fields	スキーマ内にフィールドが見つからない場合にエラーを発生させる代わりに、デフォルト値を使用するかどうか。	0
input_format_avro_null_as_default	NULL を許容しないカラムに null 値を挿入する際にエラーを発生させる代わりに、デフォルト値を使用するかどうか。	0
format_avro_schema_registry_url	Confluent Schema Registry の URL。Basic 認証を利用する場合は、URL エンコードした認証情報を URL のパス部分に直接含めることができます。

例

スキーマレジストリの使用

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;

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