PySpark 대규모 데이터 품질 검증 — Deequ로 "조용히 틀린 데이터" 잡기

① 단위 테스트   : 변환 "로직"이 맞는가 (작은 합성 데이터)  ← 별도 글 "PySpark 코드 테스트"
② 데이터 검증   : 실제 "데이터"가 기대 범위인가 (프로덕션 데이터)  ← 이 글

from pyspark.sql import functions as F
 
def validate(df):
    total = df.count()
    checks = {}
 
    # 행 수 (급감/급증 감지)
    checks["row_count"] = total
 
    # NULL 비율 (핵심 컬럼)
    checks["user_id_null_pct"] = df.filter(F.col("user_id").isNull()).count() / total
 
    # 유일성 (PK 중복)
    checks["user_id_unique"] = df.select("user_id").distinct().count() == total
 
    # 값 범위 (음수 매출)
    checks["negative_amount"] = df.filter(F.col("amount") < 0).count()
 
    # 허용 값 집합 (enum)
    valid_status = ["active", "inactive", "pending"]
    checks["invalid_status"] = df.filter(~F.col("status").isin(valid_status)).count()
 
    return checks

metrics = df.agg(
    F.count("*").alias("rows"),
    F.sum(F.col("user_id").isNull().cast("int")).alias("uid_nulls"),
    F.sum((F.col("amount") < 0).cast("int")).alias("neg_amount"),
    F.countDistinct("user_id").alias("uid_distinct"),
    F.min("amount").alias("amount_min"),
    F.max("amount").alias("amount_max"),
).collect()[0]

from pydeequ.checks import Check, CheckLevel
from pydeequ.verification import VerificationSuite
 
check = (Check(spark, CheckLevel.Error, "data quality")
    .hasSize(lambda s: s > 1_000_000)            # 행 수 하한
    .isComplete("user_id")                        # NULL 없음
    .isUnique("user_id")                          # 유일
    .isNonNegative("amount")                      # 음수 없음
    .isContainedIn("status", ["active", "inactive", "pending"])
    .hasCompleteness("email", lambda c: c > 0.95))  # 95% 이상 채워짐
 
result = (VerificationSuite(spark)
    .onData(df)
    .addCheck(check)
    .run())

# 과거 메트릭 이력과 비교 (메트릭을 테이블에 적재해두고)
history = spark.table("quality.metrics_history")
 
today_rows = metrics["rows"]
baseline = (history
    .filter("metric = 'rows'")
    .agg(F.avg("value").alias("avg"), F.stddev("value").alias("std"))
    .collect()[0])
 
# 평균에서 3σ 이상 벗어나면 경보
if abs(today_rows - baseline["avg"]) > 3 * baseline["std"]:
    raise ValueError(f"행 수 이상: {today_rows} (기대 {baseline['avg']:.0f}±{baseline['std']:.0f})")

def quality_gate(df):
    result = run_checks(df)
    if result.has_errors():
        # 나쁜 데이터를 격리하고 하류 적재를 막음
        df.write.mode("overwrite").save("quarantine/...")
        alert(result)                       # 알림
        raise QualityGateError(result)      # 파이프라인 중단
 
    df.writeTo("analytics.events").append()  # 통과 시에만 적재

[변환] → [품질 게이트] ──통과──> [프로덕션 테이블]
              │
              └──실패──> [격리 + 알림 + 중단]