AvroConfluent

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

Каждое сообщение использует формат передачи данных Confluent: магический байт (0x00), затем 4-байтовый идентификатор схемы в порядке байтов big-endian, после чего следует двоичное значение Avro. При чтении ClickHouse определяет идентификатор схемы, обращаясь к реестру. При записи ClickHouse регистрирует схему, полученную из выходных столбцов, и добавляет полученный идентификатор в начало каждой строки. Для оптимальной производительности схемы кэшируются.

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

• bytes is default, controlled by setting 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

Чтобы читать 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;

Чтобы записывать сообщения 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');

Если для вашего реестра схем требуется базовая аутентификация (например, если вы используете 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