AvroConfluent

Входные данные	Выходные данные	Псевдоним
✔	✗

Описание

Apache Avro — это ориентированный на строки формат сериализации, который использует двоичное кодирование для эффективной обработки данных. Формат AvroConfluent поддерживает декодирование однообъектных сообщений Kafka, закодированных в формате Avro, сериализованных с использованием Confluent Schema Registry (или совместимых с API сервисов).

Каждое сообщение Avro встраивает идентификатор схемы, который ClickHouse автоматически разрешает, запрашивая настроенный реестр схем. После разрешения схемы кэшируются для оптимальной производительности.

Соответствие типов данных

Таблица ниже показывает все типы данных, поддерживаемые форматом Apache Avro, и их соответствующие типы данных ClickHouse в запросах INSERT и SELECT.

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 неявно принимает null в качестве значения поля, поэтому, например, Avro union(T1, T2, null) будет преобразован в Variant(T1, T2). В результате, при создании Avro из ClickHouse, мы всегда должны включать тип null в набор типов Avro union, так как мы не знаем, является ли какое-либо значение фактически null во время вывода схемы.

*** Логические типы Avro

Не поддерживаемые логические типы данных Avro:

• time-millis
• time-micros
• duration

Настройки формата

Настройка	Описание	По умолчанию
input_format_avro_allow_missing_fields	Использовать ли значение по умолчанию вместо генерации ошибки, когда поле не найдено в схеме.	0
input_format_avro_null_as_default	Использовать ли значение по умолчанию вместо генерации ошибки при вставке значения null в ненулевую колонку.	0
format_avro_schema_registry_url	URL реестра схем Confluent. Для базовой аутентификации учетные данные можно включить непосредственно в путь URL.

Примеры

Использование реестра схем

Чтобы прочитать топик Kafka, закодированный в формате 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;

Использование базовой аутентификации

Если ваш реестр схем требует базовой аутентификации (например, если вы используете Confluent Cloud), вы можете предоставить закодированные в URL учетные данные в настройке format_avro_schema_registry_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