[DB] 데이터브릭스 CT 기반 데이터 적재(SQL Sever)

데이터브릭스 CT 기반 실시간 데이터 적재 및 최적화 전략

1. 현대적 데이터 파이프라인의 핵심: 메달리온 아키텍처

데이터브릭스에서 추천하는 메달리온 아키텍처(Medallion Architecture)는 데이터를 단계적으로 정제하여 신뢰할 수 있는 데이터셋을 구축하는 구조입니다. CT 스트리밍은 이 구조의 입구(Ingestion) 역할을 수행합니다.

* 메달리온 아키텍처란? 데이터 레이크하우스(Data Lakehouse) 환경에서 데이터를 체계적으로 조직하고, 품질을 점진적으로 향상시키기 위해 사용되는 3단계 데이터 디자인 패턴으로 데이터를 품질에 따라서, 올림픽 메달처럼 금은동으로 나눠서 저장

① Bronze (Raw Layer): 원천의 복제본

• 역할: 소스 DB에서 넘어온 변경 이력을 가공 없이 그대로 적재합니다.
• 특징: SYS_CHANGE_OPERATION(I, U, D)과 SYS_CHANGE_VERSION 정보를 포함하여, 데이터가 "어떻게 변했는지"에 대한 모든 기록을 보유합니다.
• 저장 방식: 주로 Append-only 방식으로 적재하여, 추후 데이터 유실이나 재처리가 필요할 때 '진실의 원천' 역할을 합니다.

② Silver (Cleansed/Enriched Layer): 현재 상태의 유지

• 역할: Bronze의 변경 이력을 분석하여, 비즈니스 엔티티(예: 고객, 주문)의 최종 상태를 테이블 형태로 유지합니다.
• 핵심 기술: Delta Lake의 MERGE INTO 명령어를 사용합니다.
• 로직: 같은 PK를 가진 데이터가 들어오면 기존 데이터를 업데이트(Update)하고, 없으면 삽입(Insert)하며, 삭제 표식이 있으면 삭제(Delete)합니다.

③ Gold (Curated Layer): 비즈니스 분석용

• 역할: Silver 테이블들을 조인하거나 집계하여 보고서, AI 모델 학습, BI 대시보드에 최적화된 데이터를 생성합니다.

---

2. 소스 DB(SQL Server)의 Change Tracking 메커니즘

CT는 데이터베이스 엔진 내부에서 동작하는 경량화된 추적 기능입니다.

내부 동작 원리

1. 동기적 추적: 트랜잭션이 커밋될 때, SQL Server는 내부 시스템 테이블에 "이 테이블의 이 PK를 가진 행이 변경되었다"는 사실과 함께 버전 번호를 기록합니다.
2. 버전 관리: DB 전체에 대해 글로벌한 버전 번호(CHANGE_TRACKING_CURRENT_VERSION)가 증가하며, 각 테이블은 자신의 마지막 변경 지점을 알고 있습니다.
3. 데이터 보관: Retention Period 설정을 통해 오래된 변경 기록은 자동으로 삭제(Auto Cleanup)됩니다.

CT vs CDC (Change Data Capture) 선택 기준

• CT: 변경된 PK만 기록합니다. "무엇이 변했나"만 알 수 있고 "어떤 값으로 변했나"는 원본 테이블을 다시 조회해서 가져와야 합니다. 부하가 매우 적습니다.
• CDC: 변경된 데이터 값 전체를 로그에서 읽어 별도의 이력 테이블에 저장합니다. 부하가 CT보다 크지만, 과거 모든 상태의 변화를 추적할 수 있습니다.
• 결론: 단순 실시간 동기화가 목적이라면 CT가 성능 면에서 압도적으로 유리합니다. CDC의 경우, PK가 없는 테이블에 적용합니다.

---

3. 데이터브릭스 스트리밍 구현 상세 (PySpark)

데이터브릭스에서 CT 데이터를 스트리밍으로 읽어올 때는 Spark Structured Streaming과 전용 커넥터를 사용합니다.

핵심 코드 구조 및 파라미터 설명

# 1. 소스 DB 연결 설정
source_options = {
    "host": "your-db-server.database.windows.net",
    "database": "TargetDB",
    "user": "databricks_sync_user",
    "password": dbutils.secrets.get(scope="sql", key="db_password"),
    "dbtable": "Sales.Orders", # 혹은 서브쿼리 사용
    "changeTracking": "true", # CT 모드 활성화
    "incremental": "true",    # 증분 적재 모드
    "isolationLevel": "READ_UNCOMMITTED" # 락 방지
}

# 2. 스트리밍 데이터프레임 생성
# readStream을 사용하면 Spark는 체크포인트를 확인하여 마지막 읽은 버전 이후부터 읽기 시작함
raw_stream_df = spark.readStream \
    .format("sqlserver") \
    .options(**source_options) \
    .load()

# 3. Micro-batch 처리 (Silver 레이어 Upsert)
def upsert_to_delta(batch_df, batch_id):
    # 동일 배치 내 중복 데이터 제거 (가장 최신 버전만 선택)
    window_spec = Window.partitionBy("OrderID").orderBy(col("SYS_CHANGE_VERSION").desc())
    latest_batch_df = batch_df.withColumn("rank", row_number().over(window_spec)) \
                              .filter("rank = 1").drop("rank")
    
    # Delta Table 객체 생성
    target_table = DeltaTable.forName(spark, "silver_db.orders")
    
    # MERGE 실행
    target_table.alias("t").merge(
        latest_batch_df.alias("s"),
        "t.OrderID = s.OrderID"
    ).whenMatchedUpdateAll(
        condition = "s.SYS_CHANGE_OPERATION IN ('U', 'I')"
    ).whenMatchedDelete(
        condition = "s.SYS_CHANGE_OPERATION = 'D'"
    ).whenNotMatchedInsertAll(
        condition = "s.SYS_CHANGE_OPERATION = 'I'"
    ).execute()

# 4. 스트리밍 시작 및 체크포인트 설정
query = raw_stream_df.writeStream \
    .foreachBatch(upsert_to_delta) \
    .outputMode("update") \
    .option("checkpointLocation", "/mnt/datalake/checkpoints/orders_sync") \
    .trigger(processingTime='1 minute') \
    .start()

---

4. 소스 DB 부하 최소화 및 성능 튜닝

운영 DB(OLTP)의 성능을 보호하는 것은 데이터 엔지니어링의 최우선 과제입니다.

① 락(Lock) 경합 해소: Snapshot Isolation

데이터브릭스가 대량의 데이터를 읽어갈 때 Shared Lock을 걸면 운영 앱의 Update가 차단됩니다.

이를 방지하기 위해 SQL Server에서 READ_COMMITTED_SNAPSHOT을 활성화해야 합니다.

• 원리: 변경 전 데이터를 TempDB의 버전 스토어에 보관하여, 읽기 작업이 락 없이도 일관된 데이터를 읽게 합니다.
• 주의: TempDB의 사용량이 증가하므로 TempDB 공간 모니터링이 필요합니다.

② 네트워크 부하 감소: Pushdown 필터링

dbtable 옵션에 테이블 이름 대신 WHERE 절이 포함된 서브쿼리를 사용하세요.

• 예시: (SELECT * FROM Orders WHERE Region = 'Asia' AND OrderDate > '2024-01-01') AS filtered_table
• 이렇게 하면 소스 DB 엔진이 먼저 필터링을 수행하여 꼭 필요한 데이터만 네트워크로 전송합니다.

③ 적절한 Trigger 설정

스트리밍이라고 해서 ProcessingTime='0 seconds'(가장 빠르게)로 설정하면 소스 DB에 쿼리가 너무 빈번하게 발생합니다.

데이터의 시급성에 따라 1분~10분 단위로 조절하는 것이 소스 DB 부하 조절에 효과적입니다.

---

5. 모니터링 및 장애 대응 (Troubleshooting)

① SQL Server Profiler & Extended Events

DB 부하가 의심될 때 Profiler를 통해 다음을 체크합니다.

• Duration: 데이터브릭스가 던지는 CHANGETABLE 쿼리의 수행 시간.
• Reads/Writes: 쿼리당 발생하는 논리적 읽기 수.
• Wait Type: LCK_M_... 대기가 발생하는지 확인. (발생 시 Snapshot Isolation 설정 재점검)

② CT Retention Gap 문제

소스 DB의 CT 보관 주기가 2일인데, 데이터브릭스 스트리밍이 장애로 3일간 중단되었다면?

• 현상: CHANGE_TRACKING_MIN_VALID_VERSION보다 체크포인트의 버전이 낮아져 에러 발생.
• 해결: 이 경우 증분 적재가 불가능하므로, 체크포인트를 초기화하고 Full Re-sync를 수행해야 합니다. 이를 방지하기 위해 보관 주기를 충분히(예: 7일) 잡는 것이 실무적인 팁입니다.

③ Spark UI 모니터링

데이터브릭스의 Streaming Tab에서 다음 지표를 주시하세요.

• Input Rate vs Process Rate: 데이터가 쌓이는 속도보다 처리 속도가 느리면 지연(Backlog)이 발생합니다.
• Batch Duration: 배치 처리 시간이 Trigger 주기보다 길어지면 클러스터 사양을 높이거나 병렬 처리를 늘려야 합니다.

---

6. 데이터 정합성 보장 전략

① 멱등성(Idempotency) 확보

스트리밍은 네트워크 장애 등으로 인해 동일한 배치가 재실행될 수 있습니다. Delta Lake의 MERGE INTO는 그 자체로 멱등성을 보장하므로, 중복 데이터가 들어와도 결과값은 항상 동일하게 유지됩니다.

② 스키마 진화(Schema Evolution)

소스 DB에 컬럼이 추가되면 스트리밍이 깨질 수 있습니다.

Delta Lake의 mergeSchema 옵션을 활용하거나, Bronze 레이어에서 cloudFiles.schemaEvolutionMode를 사용하여 유연하게 대응할 수 있습니다.

---

7. 결론 및 향후 과제

데이터브릭스와 SQL Server CT를 결합한 구조는 저부하, 고성능, 고가용성을 모두 만족하는 설계입니다.

특히 IT 운영 환경(SAP, SCM 등)에서 발생하는 대량의 트랜잭션 데이터를 실시간으로 분석 환경으로 가져올 때 가장 권장되는 패턴입니다.