Kafka 및 CDC 통합

요약 Apache Doris는 세 가지 경로를 통해 Kafka 토픽과 CDC 변경 스트림을 기본적으로 수집합니다. FE 내부에서 실행되며 외부 서비스가 필요 없는 Kafka 컨슈머를 원한다면 Routine Load를 사용하세요. MySQL/Postgres/Oracle 데이터베이스 전체를 동기화하면서 Stream Load 2PC를 통해 exactly-once를 보장하려면 Flink CDC와 함께 Flink Doris Connector를 사용하세요. 이미 Kafka Connect를 운영 중이고 페이로드가 Avro, Protobuf 또는 Debezium 형식이라면 Doris Kafka Connector를 사용하세요. 세 경로 모두 동일한 Unique Key Merge-on-Write 테이블에 데이터를 적재하므로, 어떤 경로를 선택하든 순서가 뒤바뀐 이벤트와 삭제는 동일하게 처리됩니다.

Apache Doris에서 Kafka 및 CDC 통합을 사용하는 이유는 무엇인가요?

Apache Doris의 Kafka 및 CDC 통합은 스트리밍 데이터 수집, exactly-once 전달, 순서가 뒤바뀐 이벤트 처리를 엔진 내부에서 수행하므로, 토픽이나 binlog의 데이터가 쿼리 가능한 행이 되기까지의 경로가 짧습니다. 대부분의 데이터 웨어하우스에서는 스트리밍 데이터 수집 기능을 별도로 구축해야 합니다. Kafka 컨슈머를 띄우고 오프셋 관리, 재시도, 스키마 매핑, 삭제 처리를 다시 구현한 뒤, 웨어하우스가 중단되었을 때 컨슈머도 문제없이 종료되기를 바라야 합니다. 또는 앞단에 Flink를 두고 추가 클러스터 비용을 부담해야 합니다. 아니면 "실시간"을 포기하고 일괄 데이터 적재로 오 분 간격의 배치를 전송해야 합니다.

• CDC 파이프라인이 체크포인트 실패 후 마지막 배치를 다시 처리하면서 버퍼에 있던 모든 행을 중복 생성합니다.
• Unique Key 테이블에 순서가 뒤바뀐 업데이트가 도착했을 때, 어떤 쓰기가 더 최신인지 엔진에 알려 주지 않아 이전 버전이 최종적으로 반영됩니다.
• Avro나 Protobuf 데이터를 담은 Kafka 토픽은 JSON이나 CSV만 지원하는 HTTP 데이터 적재 API와 호환되지 않습니다.

Doris는 이러한 처리의 대부분을 엔진 내부로 옮깁니다. 따라서 Kafka 토픽이나 MySQL binlog의 데이터가 쿼리 가능한 행이 되기까지의 경로가 짧고, 어떤 유형의 장애가 발생했는지도 명확히 파악할 수 있습니다.

Apache Doris의 Kafka 및 CDC 통합이란 무엇인가요?

Apache Doris의 Kafka 및 CDC 통합은 공통 기반을 공유하는 스트리밍 및 CDC 데이터 수집 경로의 집합입니다. 이 기반은 Label로 식별되는 트랜잭션 데이터 적재, upsert와 삭제를 위한 Unique Key Merge-on-Write 테이블, 이벤트가 순서대로 도착하지 않을 때 최종 반영할 버전을 결정하는 시퀀스 컬럼으로 구성됩니다. 각 경로는 컨슈머를 누가 실행하느냐에 따라 구분됩니다.

주요 용어

• Routine Load: FE가 관리하는 장기 실행 Kafka 컨슈머입니다. CREATE ROUTINE LOAD 문 하나를 실행하면 FE가 BE에 태스크를 할당하고, BE는 배치를 가져와 각 배치를 트랜잭션 데이터 적재로 처리합니다.
• Stream Load 2PC: PRECOMMITTED 상태에서 대기하다가 후속 호출로 txn_operation: commit 또는 txn_operation: abort를 받는 HTTP 데이터 적재입니다. 이처럼 두 단계로 나누기 때문에 외부 sink에서 exactly-once를 구현할 수 있습니다.
• Flink Doris Connector: Stream Load 2PC를 실행하고 커밋을 Flink 체크포인트와 연계하는 외부 Flink sink입니다. Flink CDC와 함께 사용하면 데이터베이스 전체를 동기화하고 DDL을 전파합니다.
• Doris Kafka Connector: Kafka Connect sink 플러그인입니다. Kafka Connect의 컨버터 생태계를 활용하므로 별도의 직렬화 구현 없이 Avro, Protobuf, Debezium 페이로드를 처리할 수 있습니다.
• function_column.sequence_col: Unique Key MOW 테이블의 테이블별 속성으로, 행의 어느 버전을 최종 반영할지 결정하는 값이 담긴 컬럼을 지정합니다. 순서가 뒤바뀐 CDC 이벤트에 대비하는 안전장치입니다.

Apache Doris의 Kafka 및 CDC 통합은 어떻게 작동하나요?

Apache Doris의 Kafka 및 CDC 통합 경로는 데이터가 FE에 도달하면 동일한 수명 주기를 따릅니다. 배치를 가져와 레이블이 지정된 트랜잭션으로 묶고, 확인을 받은 뒤 커밋한 다음 모든 복제본에 게시합니다.

1. 컨슈머가 배치를 가져옵니다. Routine Load는 BE에 태스크를 할당하여 Kafka에서 데이터를 가져옵니다. max_batch_rows(기본값 20M), max_batch_size(기본값 1 GB), max_batch_interval(기본값 60 s) 중 먼저 도달하는 한도에서 배치를 마칩니다. Flink CDC는 청크 방식의 HTTP Stream Load로 행을 스트리밍하며, 이 연결은 체크포인트 구간 내내 열린 상태를 유지합니다.
2. 배치가 트랜잭션이 됩니다. BE는 Label을 지정하여 트랜잭션을 엽니다. Routine Load는 레이블을 생성하고, Flink Doris Connector는 체크포인트 ID로 레이블을 지정합니다. 어느 경로든 동일한 레이블로 다시 제출하면 중복이 제거됩니다.
3. 커밋은 확인을 기다립니다. Routine Load는 배치가 도착하면 즉시 커밋하고, 성공한 경우에만 Kafka 오프셋을 앞으로 이동합니다. Flink CDC는 PRECOMMITTED 상태에서 멈춥니다. 다음 체크포인트 배리어가 txn_operation: commit을 트리거하고, 체크포인트가 실패하면 abort를 트리거하여 임시 저장된 데이터를 버립니다.
4. 게시하면 행을 조회할 수 있습니다. 백그라운드 데몬이 모든 복제본에 PublishVersion을 전달합니다. 트랜잭션이 VISIBLE 상태가 되면 새 행을 쿼리할 수 있습니다. 동일한 트랜잭션에 속한 Tablet은 일부만 공개되는 일이 없습니다.
5. 오류가 발생하면 데이터는 유지하고 작업을 일시 중지합니다. Routine Load 배치가 max_error_number(기본값 0)를 초과하면 작업이 PAUSED 상태로 전환되고 FE에 마지막 오류 URL이 보관됩니다. 원본 데이터를 수정한 뒤 작업을 재개하세요. 오프셋은 기존 위치에 그대로 유지됩니다.

빠른 시작

CREATE TABLE events (
  user_id BIGINT, event_type STRING, ts DATETIME
)
DUPLICATE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 8;

CREATE ROUTINE LOAD events_kafka ON events
COLUMNS(user_id, event_type, ts)
PROPERTIES (
  "format" = "json",
  "jsonpaths" = "[\"$.user_id\",\"$.event_type\",\"$.ts\"]"
)
FROM KAFKA (
  "kafka_broker_list" = "localhost:9092",
  "kafka_topic" = "events",
  "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

예상 결과

+--------+--------------+---------+----------------------+
| Id     | Name         | State   | Statistic            |
+--------+--------------+---------+----------------------+
| 100123 | events_kafka | RUNNING | {"loadedRows":4821,..|
+--------+--------------+---------+----------------------+

SHOW ROUTINE LOAD FOR events_kafka는 작업이 RUNNING 상태임을 보여 주며, 파티션별로 마지막으로 소비한 오프셋과 적재된 행 수를 보고합니다. PAUSE, RESUME, STOP은 다른 Doris 작업에서와 동일하게 작동합니다.

CDC를 사용하려면 컨슈머를 Flink CDC와 함께 사용하는 Flink Doris Connector로 바꾸세요(Flink 작업 하나, MySQL 소스, Doris sink, sink.enable-2pc = true). 스키마와 시퀀스 컬럼은 Doris 쪽에 정의합니다.

CREATE TABLE orders (
  order_id BIGINT, status STRING, updated DATETIME
)
UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 8
PROPERTIES ("function_column.sequence_col" = "updated");

sequence_col 설정은 지연된 CDC 이벤트가 최신 이벤트보다 늦게 도착했을 때 데이터를 보호합니다. 더 오래된 updated 값은 반영되지 않고 최신 값이 반영되므로, 도착 순서에 관계없이 결과가 동일합니다.

Apache Doris의 Kafka 및 CDC 통합은 언제 사용해야 하나요?

스트리밍 이벤트나 업스트림 데이터베이스의 변경 사항을 오 분 간격의 배치 대신 exactly-once 전달과 올바른 순서 처리를 보장하면서 Doris에 적재해야 할 때 Apache Doris의 Kafka 및 CDC 통합을 사용하세요.

적합한 경우

• CSV나 JSON을 담은 Kafka 토픽을 별도 서비스 없이 Doris에 적재하려는 경우. Routine Load가 가장 짧은 경로입니다.
• MySQL, Postgres 또는 Oracle의 CDC 데이터를 exactly-once 전달과 DDL 전파를 보장하면서 Doris에 적재하려는 경우. Flink CDC와 함께 Flink Doris Connector를 사용하세요.
• 이미 Kafka Connect를 운영 중이고 페이로드가 Avro, Protobuf 또는 Debezium 형식인 경우. Doris Kafka Connector를 사용하면 형식 불일치 문제를 피할 수 있습니다.
• 순서가 뒤바뀐 쓰기를 올바르게 처리해야 하는 Unique Key 테이블 기반의 실시간 대시보드. 위의 어느 경로든 function_column.sequence_col과 함께 사용하세요.

적합하지 않은 경우

• Routine Load를 통해 Kafka의 Avro 데이터를 가져오는 경우. Routine Load는 CSV와 JSON만 읽습니다. 업스트림에서 변환하거나, 적절한 컨버터와 함께 Doris Kafka Connector를 사용하세요.
• 마이크로서비스에서 이벤트마다 INSERT INTO ... VALUES를 실행하는 경우. 이는 Kafka나 CDC로 해결할 문제가 아닙니다. BE가 서버 측에서 작은 쓰기를 배치로 묶을 수 있도록 Group Commit을 사용하세요.
• Kafka 보존 기간이 짧은 토픽에 Routine Load를 사용하는 경우. 보존 기간이 24 h인데 작업이 이틀 동안 일시 중지되어 있으면, 저장된 오프셋이 사용 가능한 가장 이른 오프셋보다 뒤처져 Kafka가 "out of range" 오류로 재개를 거부합니다. 보존 기간을 늘리거나 버퍼링 기능이 있는 커넥터를 실행하세요.
• FE만으로 오프셋 지연을 모니터링하려는 경우. FE는 현재 지연 메트릭을 제공하지 않습니다(미해결 이슈). Kafka 측의 컨슈머 지연 exporter에서 메트릭을 수집하거나 BE 측 처리율을 모니터링하세요.
• 클라우드 / 스토리지와 컴퓨팅을 분리한 배포 환경에서 Merge-on-Write Unique Key 테이블에 Stream Load 2PC를 사용하는 경우. BE가 이 조합을 거부하므로, 해당 환경에서는 Flink Connector의 exactly-once 경로가 Label 기반 중복 제거와 시퀀스 컬럼을 사용하는 방식으로 대체됩니다. 온프레미스 클러스터에서는 MoW에 대해 전체 Stream Load 2PC 경로를 그대로 사용할 수 있습니다.

성능 및 수치

"Flink의 Apache Doris 실시간 쓰기가 높은 속도와 exactly-once를 모두 유지하는 방법"(2022년 칠월 릴리스 블로그)에 따르면, Stream Load를 30개 동시에 실행했을 때 종단 간 지연 시간은 일 초 미만을 유지했습니다. Flink 태스크 20개는 초당 약 100,000개 이벤트의 업스트림 처리율을 유지했으며, 높은 동시성 조건에서 BE의 CPU 사용량은 이전 데이터 적재 경로보다 약 25% 감소했습니다. 출처: Apache Doris 블로그.

더 읽어보기

• Routine Load 레퍼런스
• Flink Doris Connector 및 Flink CDC
• Avro, Protobuf, Debezium용 Doris Kafka Connector
• 데이터 적재 트랜잭션: 2PC의 기반 모델
• 데이터 업데이트 및 삭제: 시퀀스 컬럼과 Merge-on-Write
• Binlog Table Stream: 반대 방향으로 작동하는 변경 피드로, Doris의 행 변경 사항을 다운스트림 컨슈머에 제공합니다.
• Iceberg: 정제된 CDC 파이프라인에서 흔히 사용하는 대상입니다. SQL을 통한 다시 쓰기로 다른 엔진의 리더도 동기화된 상태를 유지할 수 있습니다.
• Unique Key: CDC sink가 데이터를 적재하는 테이블 모델입니다. 소스의 LSN/커밋 타임스탬프를 사용하는 시퀀스 컬럼과 함께 사용하세요.