광고 정산 데이터 Databricks 이관기 — Hadoop 배치 실패 없는 파이프라인 만들기

참고: 이 글의 코드는 실제 업무에서 경험한 내용을 바탕으로 개념적으로 재작성한 것입니다. 실제 회사 코드와는 무관합니다.

파이프라인 구조

들어가며

광고 정산 데이터는 돈이 오가는 데이터입니다. 1원의 오차도 광고주와의 신뢰 문제로 이어질 수 있습니다. 그런데 그 중요한 데이터를 처리하는 배치가 새벽마다 실패하고 있었습니다.

Gmarket AdTech팀에서 CPC 광고 Bill/Pay 정산 데이터와 랭킹 갱신 스냅샷 데이터를 Hue/Hadoop 기반에서 Databricks로 이관한 과정을 공유합니다.

문제 상황: 새벽 수동 재실행의 일상화

기존 환경

Bill/Pay 정산 데이터: Hue/Hadoop 기반 배치로 처리
랭킹 스냅샷 데이터: MongoDB 집계로 별도 처리
두 데이터의 결합: 별도 배치에서 수동으로 조율

문제는 Hadoop 인프라의 리소스 부족이었습니다. 다른 팀의 배치들과 클러스터 자원을 공유하다 보니, 특정 시간대에 광고 정산 배치가 자원을 확보하지 못하고 타임아웃으로 실패하는 일이 빈번했습니다.

그리고 배치가 실패하면 결국 사람이 새벽에 일어나 수동으로 재실행해야 했습니다.

회사 차원의 전환 결정

때마침 회사에서 Databricks 도입 결정을 내리고 Hadoop 기반 처리를 점진적으로 폐기하는 기조가 생겼습니다. 광고팀 파이프라인도 이 흐름에 맞춰 이관이 필요했습니다.

이관의 핵심 과제: 정산 데이터 정확도 검증

단순히 같은 로직을 Databricks로 옮기는 것은 어렵지 않습니다. 진짜 어려운 것은 이관 결과가 기존 결과와 정확히 일치하는지 검증하는 것이었습니다.

광고 정산 데이터 특성상 다음 3가지가 반드시 일치해야 했습니다:

건수 일치: 처리된 광고 클릭/노출 건수가 같은가
금액 합계 일치: 과금/차감 금액 총합이 같은가
키 단위 일치: 광고주별, 상품별, 기간별로 동일한 결과인가

검증 전략

[MS-SQL 원천 정산 데이터]
         ↓
[Databricks 파이프라인]
    ↓           ↓
[중간 집계]   [최종 결과]
    ↓           ↓
[기존 Hadoop 결과와 비교 검증]

단계별로 검증 체크포인트를 만들어, 문제가 발생하면 어느 단계에서 차이가 생겼는지 즉시 추적 가능하도록 설계했습니다.

파이프라인 재구성

1단계: 기존 흐름 완전 이해

이관 전 가장 많은 시간을 쏟은 것이 바로 이 단계였습니다.

Hue 콘솔에서 기존 Hadoop 잡의 SQL 쿼리들 분석
MS-SQL 원천 테이블들의 관계와 데이터 특성 파악
MongoDB 랭킹 스냅샷 데이터가 정산 계산에 어떻게 사용되는지 추적
과금과 차감 데이터의 처리 순서와 의존 관계 정리

기존 코드가 문서화되어 있지 않아, 실제 쿼리를 실행하며 역공학으로 흐름을 재구성했습니다.

2단계: Databricks 파이프라인 구성

# 원천 데이터 적재
raw_billing = spark.read \
    .format("jdbc") \
    .option("url", mssql_url) \
    .option("dbtable", "CPC_BILLING_RAW") \
    .load()

# 중간 집계 테이블 생성
agg_by_ad = raw_billing \
    .groupBy("ad_id", "advertiser_id", "date") \
    .agg(
        F.sum("charge_amount").alias("total_charge"),
        F.count("*").alias("click_count")
    )

# 검증: 중간 집계 단계
validate_checkpoint(agg_by_ad, expected_total_charge)

# 최종 정산 결과 생성
final_bill = agg_by_ad \
    .join(advertiser_budget, on="advertiser_id") \
    .withColumn("net_charge", F.least("total_charge", "remaining_budget"))

3단계: 단계별 검증 구조

def validate_checkpoint(df, checkpoint_name):
    result = df.agg(
        F.sum("total_charge").alias("sum_charge"),
        F.count("*").alias("row_count")
    ).collect()[0]
    
    expected = get_expected_from_legacy(checkpoint_name)
    
    if abs(result.sum_charge - expected.sum_charge) > TOLERANCE:
        raise DataValidationError(
            f"[{checkpoint_name}] 금액 불일치: "
            f"actual={result.sum_charge}, expected={expected.sum_charge}"
        )
    
    log.info(f"[{checkpoint_name}] 검증 통과: {result.row_count}건, {result.sum_charge}원")

이 구조 덕분에 이관 초기에 발견한 여러 집계 오차들(예: 시간대 변환 로직 차이, 소수점 처리 방식 차이)을 정확한 위치에서 빠르게 잡을 수 있었습니다.

이관 결과

Databricks로 이관 후:

배치 실패 0건: 전용 클러스터 자원으로 리소스 경합 없음
새벽 수동 재실행 완전 해소: 담당자 야간 대기 부담 제거
단계별 검증으로 정산 데이터 신뢰도 향상: 집계 오차 자동 감지

광고 정산성 데이터에서 중요한 것은 속도보다 정확도와 안정성입니다. 이 이관 작업을 통해 둘 다를 확보할 수 있었습니다.

마치며: 레거시 이관 시 가장 중요한 것

데이터 파이프라인 이관 작업을 하면서 얻은 가장 큰 교훈은, 검증 구조 없는 이관은 이관이 아니라 도박이라는 것입니다.

특히 정산성 데이터처럼 돈과 직결된 데이터는, 새 시스템의 결과가 기존과 동일함을 수학적으로 증명할 수 있어야 합니다. 그 검증 코드를 작성하는 데 전체 이관 공수의 절반 이상을 쏟았지만, 그것이 올바른 투자였습니다.