CDC & 실전 체크리스트

​

CDC 기반 증분 처리: APPLY CHANGES

​

APPLY CHANGES가 해결하는 문제

• 중복 제거: 같은 키에 대한 여러 변경 레코드 중 최신 것만 적용
• 순서 보장: 네트워크 지연 등으로 레코드가 순서대로 도착하지 않을 때 올바른 순서로 적용
• Late-arriving 데이터: 뒤늦게 도착한 과거 레코드를 정확히 처리
• SCD 유형 관리: Type 1(덮어쓰기) vs Type 2(이력 유지) 선택

참고 APPLY CHANGES는 내부적으로 Streaming Table 위에서 동작하지만, 대상 테이블을 자동으로 MERGE 하므로 append-only가 아닌 upsert 가 가능합니다.

​

SCD Type 1: 현재 상태만 유지

​

SQL 예시

-- 대상 테이블 정의
CREATE OR REFRESH STREAMING TABLE customers;

-- CDC 적용 (SCD Type 1)
APPLY CHANGES INTO customers
FROM STREAM(raw_customer_cdc)
  KEYS (customer_id)
  SEQUENCE BY updated_at
  COLUMNS * EXCEPT (_rescued_data)
  STORED AS SCD TYPE 1;

​

Python 예시

import dlt

dlt.create_streaming_table("customers")

dlt.apply_changes(
    target="customers",
    source="raw_customer_cdc",
    keys=["customer_id"],
    sequence_by="updated_at",
    except_column_list=["_rescued_data"],
    stored_as_scd_type=1
)

​

동작 설명

​

SCD Type 2: 변경 이력 유지

​

SQL 예시

-- 대상 테이블 정의
CREATE OR REFRESH STREAMING TABLE customers_history;

-- CDC 적용 (SCD Type 2)
APPLY CHANGES INTO customers_history
FROM STREAM(raw_customer_cdc)
  KEYS (customer_id)
  SEQUENCE BY updated_at
  COLUMNS * EXCEPT (_rescued_data)
  STORED AS SCD TYPE 2;

​

SCD Type 2 결과 테이블 구조

-- 2025년 9월 기준 고객 A의 주소
SELECT * FROM customers_history
WHERE customer_id = 'A'
  AND __START_AT <= '2025-09-01'
  AND (__END_AT > '2025-09-01' OR __END_AT IS NULL);
-- 결과: Busan

​

SCD Type 2 + 특정 컬럼만 추적

APPLY CHANGES INTO customers_history
FROM STREAM(raw_customer_cdc)
  KEYS (customer_id)
  SEQUENCE BY updated_at
  COLUMNS * EXCEPT (_rescued_data)
  STORED AS SCD TYPE 2
  TRACK HISTORY ON (address, membership_level);

​

APPLY CHANGES의 장점 요약

​

실전 체크리스트

​

파이프라인 설계 시

• ST vs MV 선택: Append-only 소스 → ST, 전체 결과를 선언적으로 관리 → MV
• Stateful 연산자 최소화: ST에서 dropDuplicates, window 등은 나중에 변경 시 Full Refresh가 필수이므로 신중하게 결정
• 소스 테이블 전제조건: MV의 소스 테이블에 Row Tracking 활성화 여부 확인
• 비결정적 함수 위치: CURRENT_DATE(), RAND() 등은 MV가 아닌 소스(ST)에서 처리
• JOIN 개수 제한: MV에서 JOIN은 2개 이하로, 3개 이상이면 중간 MV 분리
• Window 함수 대안: 가능하면 GROUP BY + 집계 함수로 대체
• CDC 패턴 결정: UPDATE/DELETE가 있는 소스 → APPLY CHANGES 사용, SCD Type 1 vs 2 결정
• Append Flow 설계: 향후 소스 추가 가능성이 있으면, 처음부터 Append Flow 패턴으로 설계

​

파이프라인 수정 시

• 변경 영향 분석: 수정하려는 내용이 Full Refresh를 유발하는지 확인 (이 가이드의 조건표 참조)
• EXPLAIN 사전 검증: MV 수정 시 EXPLAIN MATERIALIZED VIEW FOR로 incremental 가능 여부 확인
• 보호 설정 확인: 프로덕션 ST에 pipelines.reset.allowed = false가 설정되어 있는지 확인
• 선택적 Refresh: 전체가 아닌 특정 테이블만 Full Refresh 필요한 경우 full_refresh_selection 사용
• 소스 만료 확인: Full Refresh가 불가피한 경우, 소스 데이터(Kafka retention, S3 lifecycle 등)가 살아있는지 확인
• 체크포인트 복구 옵션 준비: Full Refresh 후 시작 지점 조정이 필요하면 modifiedAfter, startingVersion 등 사전 설정

​

모니터링

• Refresh 유형 확인: Pipeline 이벤트 로그에서 각 MV의 refresh 유형(incremental/full)을 주기적으로 확인
• 비용 모니터링: 예상보다 높은 DBU 소비가 발생하면 full recompute 발생 여부 확인
• 처리 시간 추적: MV 업데이트 시간이 갑자기 증가하면 full recompute로 전환되었을 가능성
• 소스 변경률 모니터링: 소스 테이블의 변경 비율이 높아지면 Enzyme이 full을 선택할 가능성 증가
• Delta 로그 보존: 소스 테이블의 Delta 로그 보존 기간이 충분한지 확인 (delta.logRetentionDuration)
• Row Tracking 상태: 새로 추가된 소스 테이블에 Row Tracking이 활성화되어 있는지 확인

​

주요 트러블슈팅

참고 파이프라인 이벤트 로그는 Pipeline UI > Events 탭에서 확인할 수 있으며, system.lakeflow.events 시스템 테이블에서 SQL로 쿼리할 수도 있습니다.