AvroConfluent

AvroConfluent 포맷 문서

입력	출력	별칭
✔	✔

​

설명

Apache Avro는 효율적인 데이터 처리를 위해 binary encoding을 사용하는 행 지향 직렬화 포맷입니다. AvroConfluent 포맷은 Confluent 스키마 레지스트리(또는 API 호환 서비스)를 사용해 Avro로 인코딩된 메시지를 읽고 쓸 수 있습니다.

각 메시지는 Confluent wire 형식을 사용합니다. 즉, 매직 바이트(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를 생성할 때는 스키마 추론 과정에서 어떤 값이 실제로 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	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	출력용: String 컬럼 중 Avro string으로 직렬화할 대상을 지정하는 정규식입니다(기본값은 bytes).

​

예시

​

Kafka에서 읽기

Avro로 인코딩된 Kafka 토픽을 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');

​

기본 인증 사용

스키마 레지스트리에서 기본 인증이 필요한 경우(예: 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 consumer의 오류를 디버깅하려면 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