근본 원인 분석

문제: 보고서의 매출 수치가 20% 높음

계보 역추적:
┌────────────────────────────────────────────────────┐
│ dashboards.revenue_summary (문제 발견)               │
│   총 매출: $1.2M (예상: $1.0M)                      │
└────────────────────────────────────────────────────┘
                    ↑
    [AGGREGATION: SUM(revenue)]
                    │
┌────────────────────────────────────────────────────┐
│ analytics.daily_sales (데이터 정상으로 보임)         │
│   행 수: 45,892                                     │
└────────────────────────────────────────────────────┘
                    ↑
    [JOIN + FILTER]
                    │
┌────────────────────────────────────────────────────┐
│ warehouse.orders (의심 지점!)                        │
│   행 수: 54,203 (예상: 45,000)                      │
│   ⚠️ 중복 레코드 감지!                               │
└────────────────────────────────────────────────────┘
                    ↑
    [ETL: 복사]
                    │
┌────────────────────────────────────────────────────┐
│ staging.raw_orders (원인 발견!)                      │
│   중복 키 제약 조건 없음                             │
│   2025-01-26: 동일 배치가 2회 로드됨                │
│   → ETL 작업이 재시도로 2회 실행됨                   │
└────────────────────────────────────────────────────┘

근본 원인:
- Airflow 작업이 일시적 네트워크 오류로 재시도됨
- 멱등성 검사 없이 데이터 재삽입
- 다운스트림으로 중복 레코드 전파

해결책:
1. staging.raw_orders에 고유 제약 조건 추가
2. ETL 작업에 멱등성 검사 구현
3. warehouse.orders에서 중복 제거
4. 영향받은 모든 다운스트림 테이블 재처리

데이터 품질 문제 디버깅 체크리스트:

□ 1. 문제 정의
  □ 영향받은 테이블/컬럼 식별
  □ 예상 값 vs 실제 값 문서화
  □ 영향받은 레코드 범위 결정 (전체? 일부?)

□ 2. 타임라인 분석
  □ 마지막으로 정상이었던 시점 확인
  □ 문제 시작 시간 식별
  □ 해당 기간의 배포/변경 검토

□ 3. 계보 조사
  □ 전체 업스트림 계보 가져오기 (depth 5+)
  □ 계보 경로에서 각 노드 식별
  □ 중간 노드들의 데이터 샘플 검사

□ 4. 이진 검색
  □ 중간 노드부터 시작
  □ 정상이면 다운스트림 검사
  □ 비정상이면 업스트림 검사
  □ 문제 노드 격리될 때까지 반복

□ 5. 근본 원인 확인
  □ 변환 로직 검토
  □ 코드 변경 이력 확인
  □ 구성 변경 검토
  □ 데이터 프로파일 비교 (이전 vs 현재)

□ 6. 영향 평가
  □ 다운스트림 계보 확인
  □ 영향받은 모든 자산 목록화
  □ SLA 영향 평가

□ 7. 수정 및 검증
  □ 수정 적용
  □ 테스트 데이터로 검증
  □ 다운스트림 재처리
  □ 모니터링 설정

from great_expectations as ge
from lineage_client import LineageAPI

class LineageAwareDataQuality:
    """계보를 활용한 데이터 품질 모니터링"""

    def __init__(self, lineage_api):
        self.lineage = lineage_api
        self.alerts = []

    def check_dataset_quality(self, dataset_name, df):
        """데이터셋 품질 검사 및 계보 기반 알림"""

        # Great Expectations로 데이터 검증
        ge_df = ge.from_pandas(df)

        # 기대값 정의
        expectations = {
            "row_count_stable": ge_df.expect_table_row_count_to_be_between(
                min_value=int(0.8 * self.historical_avg(dataset_name)),
                max_value=int(1.2 * self.historical_avg(dataset_name))
            ),
            "no_nulls_in_key": ge_df.expect_column_values_to_not_be_null(
                column="primary_key"
            ),
            "valid_values": ge_df.expect_column_values_to_be_in_set(
                column="status",
                value_set=["active", "inactive", "pending"]
            )
        }

        # 실패한 검사 처리
        for check_name, result in expectations.items():
            if not result.success:
                # 계보를 사용하여 영향 분석
                impact = self.lineage.get_downstream_impact(dataset_name)

                alert = {
                    "dataset": dataset_name,
                    "check": check_name,
                    "severity": self.calculate_severity(impact),
                    "affected_downstream": len(impact.datasets),
                    "critical_systems": impact.critical_count,
                    "recommended_action": self.suggest_action(check_name, impact)
                }

                self.alerts.append(alert)
                self.notify_teams(alert)

    def calculate_severity(self, impact):
        """영향 기반 심각도 계산"""
        if impact.critical_count > 0:
            return "CRITICAL"
        elif impact.datasets_count > 10:
            return "HIGH"
        elif impact.datasets_count > 3:
            return "MEDIUM"
        else:
            return "LOW"

    def suggest_action(self, check_name, impact):
        """권장 조치 제안"""
        if check_name == "row_count_stable":
            # 업스트림 계보 확인
            upstream = self.lineage.get_upstream_lineage(impact.dataset)
            return f"업스트림 소스 확인: {upstream.immediate_sources}"
        elif check_name == "no_nulls_in_key":
            return "ETL 로직의 조인 조건 검토"
        else:
            return "데이터 프로파일링 실행 및 계보 추적"

-- 양호한 기준선과 현재 데이터 비교
WITH baseline AS (
    SELECT * FROM customers_backup_20250125
),
current AS (
    SELECT * FROM customers
),
differences AS (
    SELECT
        COALESCE(b.customer_id, c.customer_id) as customer_id,
        CASE
            WHEN b.customer_id IS NULL THEN 'ADDED'
            WHEN c.customer_id IS NULL THEN 'DELETED'
            WHEN b.email != c.email THEN 'MODIFIED'
            ELSE 'UNCHANGED'
        END as change_type,
        b.email as baseline_email,
        c.email as current_email
    FROM baseline b
    FULL OUTER JOIN current c ON b.customer_id = c.customer_id
)
SELECT
    change_type,
    COUNT(*) as count,
    COUNT(*) * 100.0 / SUM(COUNT(*)) OVER () as percentage
FROM differences
GROUP BY change_type;

-- 결과:
-- change_type  | count  | percentage
-- UNCHANGED    | 44,892 | 97.8%
-- MODIFIED     | 523    | 1.1%
-- ADDED        | 312    | 0.7%
-- DELETED      | 165    | 0.4%
-- ⚠️ MODIFIED 비율이 비정상적으로 높음 → 조사 필요

# 특정 날짜의 계보 스냅샷 비교
lineage_jan25 = api.get_lineage(
    dataset="analytics.revenue",
    as_of="2025-01-25T00:00:00Z"
)

lineage_jan26 = api.get_lineage(
    dataset="analytics.revenue",
    as_of="2025-01-26T00:00:00Z"
)

# 계보 변경 감지
diff = compare_lineage(lineage_jan25, lineage_jan26)

print("계보 변경 사항:")
for change in diff.changes:
    print(f"- {change.type}: {change.description}")

# 출력:
# - NEW_DEPENDENCY: analytics.revenue now depends on warehouse.refunds
# - MODIFIED_TRANSFORM: revenue calculation changed from SUM(amount) to SUM(amount - refund_amount)
# - REMOVED_DEPENDENCY: no longer depends on staging.adjustments

문제: 월간 매출 보고서의 숫자가 지난주부터 일관성 없음

1단계: 증상 식별
- 영향: 경영진 대시보드
- 발생: 2025-01-20부터
- 패턴: 일부 날짜만 영향, 다른 날짜는 정상

2단계: 계보 로드
source_systems.orders → staging.order_events → warehouse.orders
→ analytics.daily_revenue → reports.monthly_revenue → dashboards.executive

3단계: 이진 검색
✓ dashboards.executive (문제 있음)
✓ reports.monthly_revenue (문제 있음)
✓ analytics.daily_revenue (일부 날짜만 문제)
✓ warehouse.orders (일부 날짜만 문제)
? staging.order_events (조사 중...)

4단계: 변경 이력
# Git 로그 확인
git log --since="2025-01-19" --until="2025-01-21" -- etl/

commit abc123:
  "Fix timezone handling in order_events ETL"
  - Changed: UTC → America/New_York for consistency
  ⚠️ 의심되는 변경!

5단계: 가설 검증
-- 타임존 변경이 일부 주문을 다른 날짜로 이동시킴
SELECT
    order_date_utc,
    order_date_ny,
    COUNT(*) as affected_orders
FROM staging.order_events
WHERE order_date_utc::date != order_date_ny::date
GROUP BY 1, 2;

-- 결과: 2,341개 주문이 다른 날짜로 분류됨 (오후 7시 이후)

6단계: 근본 원인 확인
- ETL 로직이 타임존 변환 추가
- 하지만 일일 집계는 여전히 UTC 날짜 사용
- 날짜 불일치로 인한 이중 계산 또는 누락

7단계: 해결
1. 일일 집계를 NY 타임존 날짜 사용하도록 업데이트
2. 2025-01-20 이후 모든 데이터 재처리
3. 타임존 처리를 문서화
4. 데이터 품질 검사 추가: 날짜별 주문 합계 검증

8단계: 영향 완화
- 다운스트림: 3개 테이블, 5개 보고서
- 재처리 시간: 4시간
- 이해관계자 알림: 경영진, 재무팀