CDC 파이프라인 구축 회고

1. 왜 시작했는가

당시 운영 DB 데이터를 분석계로 가져오는 작업은 대부분 배치로 운영되고 있었다. 하루 두 번 실행되는 작업도 있었고, 한 시간 단위로 실행되는 작업도 있었다. 이 정도 주기에서는 기존 구조로도 큰 문제가 없었다.

하지만 시간이 지나면서 기존 배치 주기로는 데이터 신선도 요구를 맞추기 어려운 경우가 늘어났다. 일부 데이터는 10분 단위, 더 나아가 그 이하의 주기로 반영되길 기대받기 시작했다. 처음에는 개별 배치의 스케줄을 조정하는 방식으로 대응할 수도 있었다. 하지만 이런 요청이 계속 늘어난다면, 단순히 배치 주기를 줄이는 방식으로는 오래 가기 어렵다고 봤다.

당시 배치는 Airflow에서 KubernetesExecutor로 실행되고 있었다. task가 실행될 때마다 Kubernetes worker pod가 생성되고, 컨테이너가 준비되고, 작업이 끝나면 pod가 종료된다. 하루 두 번 또는 한 시간 단위 배치에서는 크게 신경 쓰이지 않던 오버헤드도, 주기를 계속 줄이기 시작하면 무시하기 어려워진다. 특히 짧은 주기의 배치에서는 실제 데이터 처리 시간보다 작업을 띄우고 정리하는 시간이 더 눈에 띌 수 있다.

더 큰 문제는 운영 DB를 직접 조회한다는 점이었다. 기존 수집 방식은 원본 테이블을 주기적으로 조회해 데이터를 가져오는 구조였다. 배치 주기를 줄이면 그만큼 운영 DB를 더 자주 읽게 된다. 데이터 신선도를 높이기 위해 운영 DB를 더 자주 조회하는 방식은 장기적으로 좋은 방향이 아니라고 생각했다.

수집 방식도 일관되지 않았다. 데이터 소스마다 수집 방식과 도구가 달랐고, 테이블 추가나 변경이 필요할 때도 각각의 방식에 맞춰 별도로 대응해야 했다. 요청이 적을 때는 크게 문제가 되지 않았지만, 대상이 늘어날수록 운영 복잡도는 계속 커질 수밖에 없었다.

2. 어떤 구조를 만들려고 했는가

1장에서 정리한 문제를 해결하려면, 원본 테이블을 주기적으로 다시 읽는 방식에서 벗어나야 했다. 배치 주기를 줄이는 것만으로는 데이터 신선도 요구에 계속 대응하기 어렵고, 운영 DB를 더 자주 조회하게 된다는 부담도 그대로 남는다.

그래서 방향은 CDC로 잡았다. 초기 목표 구조는 단순하게 보면 다음과 같았다.

	  운영 DB → Source Layer → Buffer Layer → Sink Layer → 분석계 저장소

Source 쪽에서는 Debezium이 MySQL binlog를 읽어 row 단위 변경 이벤트를 Kafka topic에 기록한다. 이벤트 value는 Avro로 직렬화되고, schema는 Schema Registry에 등록된다. Kafka는 source와 sink 사이의 버퍼 역할을 한다. source가 변경 이벤트를 생산하는 속도와 sink가 BigQuery에 반영하는 속도를 분리하고, 장애가 발생했을 때도 offset과 checkpoint를 기준으로 재처리할 수 있게 하기 위한 선택이었다.

내가 맡은 부분은 Sink Layer였다. Kafka에 쌓인 변경 이벤트를 읽고, BigQuery에 반영하는 영역이다. 여기서는 Spark Structured Streaming을 사용했다. Spark를 선택한 이유는 micro-batch 단위로 처리 흐름을 제어하기 좋았고, BigQuery에 쓰기 전 필요한 변환과 중복 제거, MERGE 로직을 직접 구현할 수 있었기 때문이다.

초기 목표는 명확했다. 운영 DB를 직접 반복 조회하지 않고, Kafka에 쌓인 변경 이벤트를 기반으로 BigQuery 테이블을 최신 상태에 가깝게 유지하는 것. 그리고 이 과정을 이후 다른 테이블에도 반복 적용할 수 있는 구조로 만드는 것이었다.

3. Spark로 CDC 이벤트를 BigQuery에 반영하기

Sink Layer는 Spark Structured Streaming으로 구현했다. Kafka topic을 계속 읽되, 실제 처리는 micro-batch 단위로 수행하는 구조였다. 이때 핵심이 된 것은 foreachBatch였다.

처음에는 Spark의 streaming API 안에서 가능한 한 많은 처리를 끝내고 싶었다. 하지만 BigQuery에 반영하는 과정은 단순 append가 아니었다. 같은 primary key에 대한 이벤트가 한 batch 안에 여러 번 들어올 수 있었고, 최종 테이블에는 batch 안에서 가장 마지막 상태만 반영해야 했다. BigQuery MERGE를 수행하기 위해 batch 결과를 임시 테이블에 먼저 적재한 뒤, 별도의 MERGE query를 실행하는 과정도 필요했다.

그래서 stream 자체는 계속 실행하되, 실제 처리는 foreachBatch 안으로 모았다. 이 구조 덕분에 Spark Structured Streaming의 연속 실행 모델을 유지하면서도, BigQuery 쓰기처럼 batch 성격이 강한 작업을 직접 제어할 수 있었다.

중복 제거는 primary key 기준으로 수행했다. 같은 key에 대해 여러 이벤트가 들어오면 source timestamp 기준으로 최신 이벤트를 우선하고, timestamp가 같은 경우 Kafka offset을 tie-breaker로 사용했다. 장애 복구나 재처리 상황에서도 같은 입력에 대해 같은 결과가 나오도록 하기 위한 선택이었다.

BigQuery 반영은 temp table을 거쳐 수행했다. Spark BigQuery connector로 batch 결과를 임시 테이블에 적재하고, 이후 BigQuery query job으로 target table에 MERGE했다. Spark는 batch 데이터를 BigQuery에 쓰는 역할을 맡고, 최종 upsert는 BigQuery 쪽에서 처리하는 구조였다.

단일 테이블 기준으로는 이 구조가 비교적 명확했다. Kafka topic 하나를 읽고, batch마다 decode와 dedup을 수행한 뒤, BigQuery target table에 MERGE한다. checkpoint를 기준으로 offset을 관리하기 때문에 장애가 나더라도 마지막으로 성공한 지점 이후부터 다시 처리할 수 있었다.

하지만 운영 환경에서는 곧 몇 가지 문제가 드러났다. 먼저 스키마 변경을 제대로 다뤄야 했다. 또한 테이블 하나당 Spark job 하나를 띄우는 구조는 테이블 수가 늘어날수록 부담이 커졌다. foreachBatch 안에서 decode, parsing, dedup, BigQuery write까지 모두 수행하다 보니 batch 크기와 데이터 변환 방식이 메모리에도 직접 영향을 줬다.

결국 1차 구현으로 Kafka-to-BigQuery CDC 흐름은 만들 수 있었지만, 운영 가능한 구조로 만들기 위해서는 스키마 변경, 테이블 확장, 메모리 문제를 차례대로 풀어야 했다.

4. Schema Evolution 대응

1차 구현에서 가장 먼저 부딪힌 문제는 schema evolution이었다.

CDC 파이프라인에서는 운영 DB의 스키마가 바뀌어도 이미 Kafka에 들어간 메시지는 사라지지 않는다. 컬럼이 추가되거나 삭제되면 Kafka에는 변경 전 schema로 직렬화된 메시지와 변경 후 schema로 직렬화된 메시지가 함께 존재할 수 있다. Spark job은 이 메시지들을 순서대로 읽어야 하고, BigQuery에는 현재 테이블 schema에 맞춰 반영해야 한다.

처음에는 Spark의 from_avro를 사용하는쪽으로 진행했다. Spark SQL 함수 안에서 Avro payload를 바로 DataFrame으로 디코딩할 수 있다면 구조가 가장 단순하다. 하지만 CDC 환경에서 필요했던 것은 단순 Avro decode가 아니었다. 메시지는 생성 시점의 writer schema로 직렬화되어 있고, 파이프라인은 현재 실행 시점의 reader schema를 기준으로 데이터를 해석해야 했다. 즉, Avro에서 권장하는 writer schema와 reader schema 기반의 schema resolution이 필요했다.

문제는 Spark의 from_avro가 이 요구와 맞지 않았다는 점이다. from_avro는 호출 시점에 전달한 schema를 기준으로 payload를 디코딩한다. CDC 메시지마다 Confluent wire format에 포함된 schema id를 읽고, 그 schema id에 해당하는 writer schema를 가져온 뒤, 현재 reader schema와 함께 DatumReader(writer, reader)를 적용하는 방식이 아니었다. 다시 말해 메시지별 writer schema와 현재 reader schema를 조합해 Avro 표준 schema resolution을 수행하는 흐름을 from_avro만으로는 제어하기 어려웠다.

이 차이는 운영 중 schema가 바뀌는 상황에서 중요했다. 예를 들어 Spark job이 최신 schema를 reader schema로 들고 재시작했더라도, Kafka에는 이전 schema로 직렬화된 메시지가 남아 있을 수 있다. 이때 필요한 것은 “최신 schema로만 디코딩”하는 것이 아니라, 각 메시지의 writer schema와 현재 reader schema를 함께 사용해 호환 가능한 방식으로 해석하는 것이다. writer에는 있지만 reader에는 없는 필드는 무시하고, reader에는 있지만 writer에는 없는 필드는 default 값으로 채우는 식의 Avro schema resolution이 필요했다.

그래서 from_avro 기반 처리는 포기하고, 직접 Avro 디코딩 경로를 만들었다. Kafka 메시지에서 Confluent wire format의 schema id를 추출하고, Schema Registry에서 해당 id의 writer schema를 조회한 뒤, 현재 reader schema와 함께 Avro DatumReader에 넘기는 방식이었다.

	datum_reader = avro.io.DatumReader(writer_schema, reader_schema)
	decoded = datum_reader.read(binary_decoder)

이 구조에서는 메시지마다 실제 writer schema를 기준으로 디코딩할 수 있다. 동시에 현재 파이프라인이 기대하는 reader schema를 함께 적용하므로, Avro 표준 resolution 규칙 안에서 schema 변경을 흡수할 수 있다.

문제는 이 로직을 Spark streaming DataFrame API 위에서 깔끔하게 처리하기 어렵다는 점이었다. 메시지마다 schema id가 다를 수 있고, partition 단위로 writer schema를 조회하고 DatumReader를 재사용하는 편이 효율적이었다. 그래서 foreachBatch 내부에서 RDD 레벨로 내려가 mapPartitions 기반으로 Avro 디코딩을 수행했다.

이 선택으로 schema evolution에 대한 호환성은 확보할 수 있었다. 하지만 대가도 있었다. DataFrame API만 사용할 때보다 구조가 복잡해졌고, RDD에서 디코딩한 결과를 다시 DataFrame으로 올리는 과정이 필요해졌다. 이 RDD ↔ DataFrame 왕복은 이후 메모리 문제를 분석할 때 다시 중요한 원인으로 등장했다.

결국 schema evolution 대응은 단순히 target table에 컬럼을 추가하는 문제가 아니었다. 이미 Kafka에 들어간 과거 메시지의 writer schema, 현재 Spark job의 reader schema, BigQuery target table schema가 함께 맞물리는 문제였다. 이 부분을 직접 제어하기 위해 Spark의 기본 from_avro 경로를 벗어나, DatumReader(writer, reader) 기반의 디코딩 경로를 구현했다.

5. 여러 테이블을 하나의 Spark job에서 처리하기

Schema evolution 대응이 어느 정도 정리된 뒤에는 테이블 수가 문제가 됐다.

초기 구조는 테이블 하나를 하나의 Spark streaming job으로 처리하는 방식에 가까웠다. 이 구조는 단순했다. Kafka topic 하나를 읽고, BigQuery table 하나에 반영한다. 장애가 나도 어떤 테이블에서 문제가 생겼는지 명확하고, checkpoint나 consumer group도 테이블 단위로 자연스럽게 분리된다.

하지만 테이블이 늘어나면 이 단순함이 그대로 비용이 된다. 테이블마다 Dataproc job이 하나씩 뜨고, job마다 driver가 필요하다. 실제 처리량이 많지 않은 테이블이라도 driver는 계속 떠 있어야 한다. CDC 대상 테이블이 늘어날수록 driver 수, 배포 단위, 모니터링 대상이 함께 늘어나는 구조였다.

그래서 구조를 아래처럼 바꿨다.

Before
  table A → Spark job A
  table B → Spark job B
  table C → Spark job C

After
  DB group → Spark job
               ├─ StreamingQuery(table A)
               ├─ StreamingQuery(table B)
               └─ StreamingQuery(table C)

테이블 단위로 Spark job을 띄우는 대신, 소스 DB 그룹 단위로 하나의 Spark job을 띄우고 그 안에서 여러 테이블을 처리하는 구조로 바꿨다. 하나의 SparkSession을 공유하고, 테이블마다 별도의 StreamingQuery를 실행하는 방식이다.

중요한 것은 모든 것을 하나로 합친 게 아니라는 점이다. driver와 SparkSession은 공유하지만, 테이블별 처리 상태는 분리해야 했다. 각 테이블은 독립적인 Kafka consumer group, checkpoint path, BigQuery writer를 가져갔다. 그래야 한 테이블의 offset이나 checkpoint 문제가 다른 테이블로 번지지 않고, 장애가 발생했을 때도 테이블 단위로 재시작할 수 있다.

이 변경으로 Spark job 수는 줄일 수 있었지만, 대신 job 내부 복잡도가 늘어났다. 이전에는 Spark job 하나가 곧 테이블 하나였기 때문에 job의 실패를 테이블 실패로 해석하면 됐다. 하지만 multi-table 구조에서는 하나의 job 안에 여러 StreamingQuery가 존재한다. 특정 테이블의 query만 실패하고 나머지는 계속 처리되는 상황을 고려해야 했다.

그래서 실패를 다루는 단위도 job이 아니라 query로 낮췄다. driver 쪽에서 각 StreamingQuery의 상태를 주기적으로 확인하고, 비정상 종료된 query가 있으면 해당 테이블의 query만 다시 시작했다. SparkSession은 유지하고, 실패한 table query만 새로 생성하는 방식이었다.

다만 실패한 query를 무조건 계속 재시작하는 것도 위험했다. 특정 테이블에서 schema 문제나 데이터 문제, BigQuery 적재 문제가 반복되면 query가 뜨자마자 다시 실패할 수 있다. 이 상태에서 제한 없이 재시작하면 driver는 같은 query를 계속 띄우고 죽이는 루프에 빠진다.

이를 막기 위해 테이블별로 실패 횟수와 시간 창을 관리했다. 일정 시간 안에 실패 횟수가 임계치를 넘으면 해당 테이블은 더 이상 자동 재시작하지 않고 격리했다. 이렇게 하면 특정 테이블 하나가 반복적으로 실패하더라도 나머지 테이블의 처리는 계속 유지할 수 있었다.

결국 이 구조의 핵심은 여러 테이블을 같이 실행하는 것이 아니었다. 같이 실행하되, offset과 checkpoint는 분리하고, 실패도 테이블 단위로 격리하는 것이었다. 리소스를 줄이기 위해 job을 합쳤지만, 장애 경계까지 합쳐버리면 안 됐다.

6. OOM과 BigQuery 적재 방식의 트레이드오프

Multi-table 구조로 바꾼 뒤 가장 크게 드러난 문제는 메모리였다.

테이블을 하나의 Spark job 안에 묶으면서 driver 수는 줄었지만, 한 job 안에서 여러 StreamingQuery가 동시에 실행되기 시작했다. 각 query는 자기 Kafka topic을 읽고, micro-batch마다 Avro decode, CDC parsing, dedup, BigQuery write를 수행했다. 단일 테이블 기준으로는 문제가 없어 보이던 처리도 여러 테이블이 같은 driver와 executor 리소스를 공유하자 부담이 커졌다.

처음에는 단순한 리소스 부족으로 봤다. driver memory를 늘리고, executor memory를 조정하고, batch당 읽는 Kafka 메시지 수를 줄였다. maxOffsetsPerTrigger를 낮추면 한 번에 처리하는 데이터 양이 줄어드니 batch 메모리 사용량도 줄일 수 있다. 실제로 단기적으로는 효과가 있었다.

하지만 이것만으로는 충분하지 않았다. 메모리를 계속 늘리는 방식은 근본적인 해결책이 아니었다. batch 크기를 줄이면 OOM 가능성은 낮아지지만, 처리량도 함께 줄어든다. 처리량이 줄면 Kafka lag가 쌓이고, 결국 데이터 신선도라는 원래 목표와 다시 충돌한다.

문제를 다시 보면서 BigQuery 적재 방식도 의심하게 됐다. 당시 Spark BigQuery connector를 통해 batch 결과를 쓰는 과정에서는 중간 저장소를 거치는 경로가 있었다. 이 과정에서 executor 메모리 사용량이 커졌고, 여러 table query가 동시에 write를 수행할 때 부담이 더 커졌다.

그래서 BigQuery 적재 방식을 direct write로 바꿨다. 중간 저장소를 거치는 부담을 줄이고, BigQuery Storage Write API 기반으로 직접 적재하는 방식이었다. 이 변경 이후 Spark 쪽 메모리 사용량과 OOM은 어느 정도 완화됐다. multi-table streaming job을 계속 운영하기 위해서는 필요한 변경이었다.

하지만 이 선택이 문제를 완전히 해결한 것은 아니었다. 중간 경로의 부담을 줄이자, 이번에는 BigQuery 리소스 사용이 더 직접적으로 드러났다. 적재 작업과 분석 쿼리가 같은 리소스를 두고 경합할 수 있다는 점도 이후 운영하면서 확인하게 됐다.

이때부터 문제를 조금 다르게 보게 됐다. 처음에는 Spark job을 어떻게 안정화할지만 고민했다. driver memory를 조정하고, executor OOM을 줄이고, batch 크기를 낮추고, write path를 바꾸는 식으로 하나씩 대응했다. 하지만 대응을 할수록 Spark Sink Layer가 너무 많은 책임을 가지고 있다는 생각이 들었다. CDC 이벤트 해석, schema evolution, dedup, BigQuery MERGE, 장애 격리, 메모리 관리까지 모두 Spark job 안에서 직접 처리하고 있었다.

Direct write는 Spark 파이프라인을 당장 안정화하는 데는 효과가 있었다. 하지만 동시에 Spark 기반 Sink 방식이 가진 구조적 한계도 더 분명하게 보여줬다. Spark의 메모리 문제를 줄이면 BigQuery 리소스 문제가 드러나고, 처리량을 줄이면 lag가 쌓이고, 처리량을 늘리면 다시 리소스 문제가 커졌다. 문제를 해결했다기보다 병목의 위치를 옮기며 균형을 맞추는 과정에 가까웠다.

결국 이 경험은 이후 Sink 구조를 다시 고민하게 만든 계기 중 하나가 됐다. Spark를 더 튜닝해서 모든 문제를 안고 가기보다는, CDC 이벤트를 더 단순한 형태로 적재하고 이후 처리 계층에서 정리하는 방향이 더 적절할 수 있겠다는 문제의식이 생겼다.

7. MERGE에서 append 중심 구조로

Direct write 전환 이후 Spark Sink는 한동안 안정적으로 동작했다. BigQuery 적재 경로를 바꾸면서 Spark 쪽 메모리 문제도 많이 줄었고, 적은 수의 테이블을 대상으로 운영할 때는 MERGE 구조도 크게 문제없이 동작했다.

초기 구조에서 MERGE는 자연스러운 선택이었다. CDC 이벤트를 BigQuery target table의 최신 상태로 바로 반영하려면, primary key 기준으로 batch 안의 이벤트를 정리한 뒤 target table에 upsert해야 한다. Spark는 batch마다 CDC 이벤트를 읽고, 같은 key에 대한 중복 변경을 정리한 뒤, BigQuery temp table에 적재하고, target table에 MERGE를 수행했다.

이 구조의 장점은 분명했다. BigQuery target table을 조회하는 쪽에서는 별도의 후처리를 크게 신경 쓰지 않아도 된다. CDC 이벤트가 들어오면 Sink Layer가 최종 상태에 가깝게 정리해주기 때문이다. 적은 수의 테이블에서는 이 방식이 충분히 단순하고 안정적으로 동작했다.

하지만 운영 대상을 더 넓게 보면 다른 문제가 보이기 시작했다. 당시 기존 배치 기반으로 운영되던 테이블 수는 약 1,000개 수준이었다. 만약 이 테이블들을 같은 방식으로 CDC 파이프라인에 이관한다면, 테이블마다 micro-batch 주기로 MERGE query job이 발생할 수 있다. 즉, Spark job 수를 줄이더라도 BigQuery 쪽에서는 테이블 수만큼 query job이 계속 만들어지는 구조가 된다.

MERGE는 결국 BigQuery query job으로 실행된다. 테이블 수가 적을 때는 문제가 되지 않지만, 대상 테이블이 수백 개에서 1,000개 수준으로 늘어나면 이야기가 달라진다. 동시에 실행되거나 대기하는 query job 수가 늘어나고, 분석 쿼리와 같은 BigQuery 리소스를 공유하게 된다. 이 구조를 기본값으로 두면, CDC 적재 작업이 BigQuery 쪽 운영 한계나 슬롯 경합에 영향을 받을 수 있다고 봤다.

반면 append는 성격이 조금 달랐다. 변경 이벤트를 최종 테이블 상태로 매번 반영하는 대신, 이벤트 자체를 계속 쌓는 방식이다. Spark Sink 입장에서는 batch마다 target table에 MERGE를 수행하지 않아도 되고, BigQuery query job을 매번 만들 필요도 없다. BigQuery Storage Write API를 통한 append 적재는 query job 중심의 MERGE보다 ingestion 경로에 가깝고, 대규모 테이블 수를 기본 처리 모델로 가져가기에는 더 적합해 보였다.

그래서 기본값을 MERGE가 아니라 append로 두는 방향을 검토했다. MERGE가 필요한 일부 테이블은 별도로 처리하더라도, 전체 CDC 파이프라인의 기본 동작은 변경 이벤트를 안정적으로 append 적재하는 쪽이 더 낫다고 봤다.

물론 append 구조는 대가가 있다. target table이 곧바로 최신 상태를 의미하지 않는다. 같은 primary key에 대한 여러 변경 이벤트가 계속 쌓이기 때문에, 조회 시점에는 최신 이벤트를 고르거나, 별도의 current view를 제공하거나, 주기적으로 compaction을 수행해야 한다. 다시 말해 MERGE를 없애는 것이 아니라, 최종 상태를 만드는 책임을 Spark Sink 밖으로 옮기는 것이다.

이 지점에서 다시 고민이 생겼다. append를 기본값으로 가져가려면 결국 후행 layer가 필요하다. raw change event를 쌓는 테이블, 최신 상태를 보여주는 view, 오래된 이벤트를 정리하는 compaction, snapshot 관리 같은 책임이 새로 생긴다. 이 책임들을 BigQuery 위에서 직접 하나씩 만들 수도 있었지만, 그렇게 하면 Spark Sink에서 내려놓은 복잡도를 다른 곳에 다시 직접 구현하는 셈이 된다.

그래서 Iceberg 기반 구조를 검토하게 됐다. CDC 이벤트를 append 형태로 쌓고, 그 위에서 snapshot과 current state를 관리하는 구조가 필요하다면, 이를 직접 구현하기보다 그런 문제를 더 자연스럽게 다룰 수 있는 테이블 포맷과 처리 구조를 쓰는 편이 낫다고 봤다.

MERGE 구조는 작은 규모에서는 잘 동작했고, 실제 운영에서도 한동안 안정적으로 사용했다. 다만 전체 운영 테이블 규모로 확장했을 때는 테이블마다 BigQuery query job을 계속 발생시키는 구조를 기본값으로 두기 어렵다고 봤다. Spark job 수를 줄이더라도, BigQuery 쪽에서는 테이블 수와 micro-batch 주기에 비례해 MERGE 작업이 계속 만들어지는 구조이기 때문이다.

그래서 기본 처리 모델은 MERGE보다 append에 가깝게 가져가는 편이 낫다고 판단했다. append는 Spark Sink가 매번 target table의 최신 상태를 직접 만들지 않아도 되게 해주고, CDC 이벤트를 안정적으로 적재하는 역할에 더 집중할 수 있게 한다. 대신 append 이후에는 current state 조회, compaction, snapshot 관리 같은 후행 책임이 생긴다.

이 지점에서 이후 구조는 Iceberg 기반으로 넘어가게 됐다. 이 글에서는 Iceberg 구조까지 자세히 다루지는 않으려고 한다. 다만 Spark Sink를 운영하면서 확인한 한 가지는 분명했다. Sink Layer가 모든 테이블의 최신 상태를 직접 만드는 구조는 확장될수록 부담이 커진다. 이후 구조에서는 CDC 이벤트 적재와 current state 관리를 분리하는 방향으로 책임을 다시 나누게 됐다.

8. 결과

운영 DB를 반복 조회하던 배치 중심 수집에서, Kafka 기반 CDC 이벤트 수집 구조로 넘어갈 수 있었다. Spark Sink Layer는 Kafka에 쌓인 변경 이벤트를 읽고, micro-batch 단위로 처리해 BigQuery에 반영하는 역할을 맡았다.

이 과정에서 schema evolution, table query 단위 checkpoint, 장애 격리, BigQuery 적재 방식 같은 기준도 함께 정리됐다. 모든 테이블을 같은 방식으로 안정적으로 운영할 수 있었던 것은 아니지만, 변경 이벤트 기반 수집 구조를 실제 운영 환경에서 검증할 수 있었다.

작은 규모에서는 MERGE 기반 구조도 안정적으로 동작했다. 다만 전체 운영 테이블 규모로 확장했을 때는 append와 후행 처리 계층으로 책임을 나누는 쪽이 더 적절하다고 판단했고, 이후 구조는 그 방향으로 이어졌다.

9. 회고

지금은 Spark로 Sink Layer를 계속 가져가고 있지는 않다. CDC 이벤트를 BigQuery target table에 직접 MERGE하는 구조보다, 변경 이벤트를 append하고 current state와 compaction 같은 후행 책임을 별도 계층에서 다루는 방향이 더 적절하다고 봤기 때문이다. 그래서 지금은 Iceberg 기반 구조에 더 집중하고 있다.

그렇다고 Spark Sink 작업이 의미 없었던 것은 아니다. 당시 목표 중 하나는 CDC Sink를 만드는 것뿐 아니라, Spark 기반 작업을 운영 환경에서 개발하고 실행할 수 있는 기반을 만드는 것이기도 했다. Dataproc 환경에서 job을 배포하고, 실행 상태를 확인하고, 로그와 지표를 보고, 실패한 작업을 다시 실행하거나 격리하는 흐름을 직접 만들었다. CDC Sink는 그 기반을 만들고 검증한 첫 번째 큰 사용 사례에 가까웠다.

또한 Spark Sink를 운영하면서 CDC 파이프라인에서 어떤 책임을 어디에 둘지 더 분명하게 볼 수 있었다. Kafka에서 이벤트를 읽고, schema evolution을 처리하고, Debezium envelope를 해석하고, BigQuery에 반영하는 것까지는 Spark 안에서 어느 정도 제어할 수 있었다. 하지만 테이블 수가 늘어나고, MERGE 작업이 많아지고, current state와 compaction 같은 후행 관리까지 고려하기 시작하면 Sink Layer가 맡는 책임이 너무 커진다.

이 경험 때문에 지금은 Iceberg를 단순한 저장 포맷으로만 보지 않게 됐다. CDC 이벤트를 append 형태로 쌓고, snapshot과 current state를 관리하며, 후행 compaction까지 고려해야 한다면 이 책임들을 어디에 둘지 설계하는 것이 중요하다. Spark Sink를 운영하면서 겪었던 문제들이 결국 Iceberg 기반 구조에서 다시 풀어야 할 질문으로 이어졌다.

돌아보면 Spark Sink는 최종 구조라기보다는 다음 구조로 넘어가기 위한 중간 단계에 가까웠다. 하지만 그 과정에서 CDC 이벤트를 운영 환경에서 다루기 위한 기준과 Spark job 운영 기반은 남았다. 지금은 그 경험을 바탕으로, 변경 이벤트를 더 안정적으로 저장하고 관리할 수 있는 Iceberg 기반 구조에 집중하고 있다.