2026년 4월 22일 수요일
글 목록
Lv.2 초급데이터 엔지니어링
25분 읽기Lv.2 초급
시리즈Data Engineering 플레이북 · 파트 4/7시리즈 허브 보기

Data Engineering 플레이북 — Part 4: 데이터 품질 & 거버넌스 심층 가이드

Data Engineering 플레이북 — Part 4: 데이터 품질 & 거버넌스 심층 가이드

데이터 거버넌스 없는 데이터 레이크는 데이터 스왐프가 된다. 이 글에서는 데이터 품질 6대 차원 프레임워크와 SLA 등급 설계, DataGovOps 4대 원칙, Data Contract 설계와 CI/CD 자동 강제, 데이터 카탈로그·계보·옵저버빌리티 통합 운영, PII 마스킹 기법, Snowflake RBAC 설계까지 2026년 데이터 품질과 거버넌스를 코드로 구현하는 모든 것을 다룬다.

시리즈 구성

  • Part 1 — 개요 & 2026 핵심 트렌드 (완료)
  • Part 2 — 데이터 아키텍처 설계 (완료)
  • Part 3 — 데이터 파이프라인 구축 실전 가이드 (완료)
  • Part 4 — 데이터 품질 & 거버넌스 심층 가이드 (현재 편)
  • Part 5 — 클라우드 & 인프라 (FinOps, IaC) (연재 예정)
  • Part 6 — AI-Native 데이터 엔지니어링 (연재 예정)
  • Part 7 — DataOps & 팀 운영 플레이북 (연재 예정)

목차

  1. 왜 데이터 품질과 거버넌스인가?
  2. 데이터 품질 6대 차원 프레임워크
  3. DataGovOps — 거버넌스를 코드로
  4. 데이터 계약 (Data Contract) — 생산자와 소비자 사이의 약속
  5. 데이터 카탈로그 구축과 운영
  6. 데이터 계보 (Data Lineage) — 데이터의 여정 추적
  7. 데이터 옵저버빌리티 — 신뢰의 인프라
  8. PII 관리 & 컴플라이언스 자동화
  9. 접근 제어 & 역할 설계
  10. 실전 체크리스트

1. 왜 데이터 품질과 거버넌스인가?

2026년 데이터 거버넌스는 선택이 아닌 전략적 차별화 요소다. AI가 내놓는 결과물의 품질은 학습 데이터의 품질에 종속된다. 잘못된 데이터 위에서 아무리 정교한 AI 모델을 돌려도 결과는 신뢰할 수 없다.

데이터 품질 저하의 실제 비용은 다음과 같다. 나쁜 데이터 품질은 대부분의 거버넌스 실패의 근본 원인이며, AI 학습 데이터의 품질이 모델 성능의 병목으로 떠오르고 있다. 2026년 전 세계적으로 복잡해진 AI 규제 환경(EU AI Act, GDPR, CCPA, 국내 개인정보보호법)에서 컴플라이언스 위반은 수천만 유로의 과태료로 이어진다.

"데이터 거버넌스 없는 데이터 레이크는 데이터 스왐프가 된다. 데이터 품질 없는 AI는 자신감 있는 실수 기계가 된다."

거버넌스 vs 컴플라이언스

두 개념은 자주 혼동되지만 목적이 다르다.

데이터 거버넌스데이터 컴플라이언스
정의데이터를 어떻게 관리·제어할 것인가에 대한 내부 프레임워크외부 법률·규정을 준수하는 것
초점정책, 역할, 품질 표준 내부 설정GDPR, HIPAA, CCPA 등 외부 요건 충족
관계거버넌스가 토대컴플라이언스는 그 결과물

거버넌스는 데이터를 어떻게 관리하는가를 설정하고, 컴플라이언스는 그 관리가 규칙 안에서 이루어지는지 확인한다.


2. 데이터 품질 6대 차원 프레임워크

데이터 품질을 "데이터가 맞는가?" 한 마디로 측정할 수는 없다. 실무에서는 6가지 차원으로 나눠 측정하고 목표를 설정한다.

차원정의측정 방법 예시
정확성 (Accuracy)실제 세계 값을 올바르게 반영하는가?랜덤 샘플 대조, 소스 시스템 비교
완전성 (Completeness)필수 필드가 모두 채워져 있는가?NOT NULL 비율, 필드 채움률(%)
유효성 (Validity)허용 형식·범위·규칙을 준수하는가?정규식 검사, 허용값 목록 확인, 범위 검사
일관성 (Consistency)시스템 간·시간에 걸쳐 값이 일관적인가?크로스 시스템 조인 검증, 이력 비교
유일성 (Uniqueness)중복 레코드가 없는가?PK 중복 카운트, 유사 레코드 탐지
시의성 (Timeliness)데이터가 충분히 최신인가?마지막 갱신 시간, 신선도(Freshness) SLA

품질 측정을 코드로: Great Expectations 예시

Great Expectations를 활용해 6대 차원을 코드로 정의하고 자동 검증할 수 있다.

import great_expectations as gx

context = gx.get_context()

datasource = context.sources.add_pandas("orders_datasource")
data_asset = datasource.add_dataframe_asset(name="orders")

suite = context.add_expectation_suite("orders_quality_suite")

# ① 정확성: order_amount는 양수여야 함
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="order_amount_usd",
        min_value=0.01,
        max_value=99999.99
    )
)

# ② 완전성: 필수 컬럼에 NULL 없어야 함
for col in ["order_id", "customer_id", "order_status", "ordered_at"]:
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column=col)
    )

# ③ 유일성: order_id는 중복 없어야 함
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# ④ 유효성: order_status는 허용 값만 가능
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="order_status",
        value_set={"placed", "shipped", "delivered", "cancelled", "refunded"}
    )
)

# ⑤ 볼륨 검사: 하루 주문이 최소 100건 이상
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(
        min_value=100,
        max_value=1_000_000
    )
)

validator = context.get_validator(
    datasource_name="orders_datasource",
    data_asset_name="orders",
    expectation_suite=suite
)
results = validator.validate()

if not results.success:
    raise DataQualityError(f"품질 검사 실패: {results.statistics}")

품질 SLA 등급 설계

모든 데이터셋에 동일한 수준의 품질 기준을 적용하면 비용이 폭발한다. 비즈니스 영향도에 따라 등급을 구분하는 것이 현실적이다.

등급대상정확성 목표신선도 SLA모니터링 주기
GoldKPI, 재무 보고, AI 학습 데이터99.9%1시간 이내실시간
Silver운영 분석, 마케팅 대시보드99.0%4시간 이내1시간
Bronze탐색적 분석, 원시 아카이브95.0%24시간 이내일별

3. DataGovOps — 거버넌스를 코드로

DataGovOps는 2026년 데이터 거버넌스의 핵심 패러다임이다. DevOps가 소프트웨어 배포를 자동화했듯, DataGovOps는 컴플라이언스 절차, 감사 추적, 데이터 계보 추적을 수동 감독이 아닌 코드와 자동화로 처리한다.

거버넌스를 스프레드시트로 관리하던 시대는 끝났다. 이제 거버넌스는 개발 워크플로우에 직접 내재화되는 1등급 엔지니어링 과제다.

DataGovOps 4대 원칙

원칙 1: 정책-as-코드 (Policy as Code) 거버넌스 규칙을 사람이 읽는 문서가 아닌 실행 가능한 코드로 정의한다. Git으로 버전 관리하고 CI/CD로 자동 적용한다.

원칙 2: 왼쪽으로 이동 (Shift Left) 품질·보안 검사를 파이프라인 끝이 아닌 시작점에서 수행한다. 소스 시스템에서 데이터가 입력되는 순간부터 검증을 시작한다.

원칙 3: 자동화 우선 (Automation First) 사람의 판단이 필요한 부분만 수동으로 남기고 나머지는 모두 자동화한다. PII 탐지, 접근 통제, 감사 로그, 계보 추적을 자동화한다.

원칙 4: 지속적 모니터링 (Continuous Monitoring) 한 번 설정하고 끝이 아닌 24/7 파이프라인 건강 상태 감시. 이상 감지 시 자동 알림과 차단이 작동해야 한다.

거버넌스 역할 정의 (RACI)

명확한 역할 없이는 거버넌스가 작동하지 않는다. 모든 데이터셋에 대해 누가 책임지는지 먼저 정의해야 한다.

역할책임
데이터 오너 (Data Owner)시니어급. 특정 데이터셋의 비즈니스 사용에 최종 책임. 접근 권한 승인, 데이터 분류 결정
데이터 스튜어드 (Data Steward)운영 담당. 메타데이터 유지, 품질 규칙 모니터링, 비즈니스 용어 정의 관리
데이터 엔지니어파이프라인 구현, 품질 검사 코드화, 계보 추적 기술 구현
데이터 제품 매니저비즈니스 사용자와 기술팀 간 연결고리. 데이터셋을 제품으로 관리
CDO (Chief Data Officer)거버넌스 전략 방향, 정책 생성, 이해관계자 조율

거버넌스 프레임워크 유형 선택

모델특징적합 조직
중앙 집중형중앙 데이터 팀이 모든 정책 관리규제 강한 금융·의료
분산형 (Data Mesh)도메인 팀이 자체 정책 관리대형 기술 기업
하이브리드 (연합형)전사 표준은 중앙, 실행은 도메인2026년 가장 인기 있는 모델

4. 데이터 계약 (Data Contract)

데이터 계약이란?

데이터 계약(Data Contract)은 데이터 생산자(Producer)와 소비자(Consumer) 사이의 공식적인 약속이다. 데이터셋이 무엇을 보장하는지 — 스키마, 신선도, 볼륨, 의미론적 의미 — 를 명문화하고 자동으로 강제한다.

2026년 데이터 계약은 이론에서 일상적인 실천으로 완전히 넘어왔다. 생산자는 데이터가 소비자에게 도달하기 전에 계약 위반 여부를 검증하고, 소비자는 스키마가 갑자기 변하거나 볼륨이 급감해도 대시보드나 모델이 망가지기 전에 감지할 수 있다.

데이터 계약은 소스 팀이 무언가를 변경해야 할 때 데이터 엔지니어링팀과 조율하도록 강제하는 커뮤니케이션 메커니즘이기도 하다.

데이터 계약 YAML 명세

# data_contract_orders.yaml

apiVersion: v1
kind: DataContract
metadata:
  name: orders
  owner: "order-domain-team@company.com"
  version: "2.1.0"
  status: active
  created_at: "2026-01-15"
  updated_at: "2026-04-01"

schema:
  fields:
    - name: order_id
      type: STRING
      nullable: false
      description: "주문 고유 식별자 (UUID)"
      pii: false

    - name: customer_email
      type: STRING
      nullable: true
      description: "고객 이메일 주소"
      pii: true                  # PII 플래그 → 자동 마스킹 트리거
      classification: SENSITIVE

    - name: order_amount_usd
      type: FLOAT64
      nullable: false
      constraints:
        min: 0.01
        max: 99999.99

    - name: order_status
      type: STRING
      nullable: false
      constraints:
        allowed_values:
          - placed
          - shipped
          - delivered
          - cancelled
          - refunded

    - name: ordered_at
      type: TIMESTAMP
      nullable: false
      description: "주문 생성 시각 (UTC)"

quality:
  freshness:
    max_age_hours: 2
    warn_after_hours: 1
  completeness:
    min_completeness_pct: 99.5
  volume:
    min_daily_rows: 500
    max_daily_rows: 5_000_000
    anomaly_threshold_pct: 30
  uniqueness:
    unique_columns:
      - order_id

slo:
  availability: 99.9
  latency_p95_seconds: 60
  incident_response_minutes: 30

consumers:
  - name: "revenue-dashboard"
    team: "analytics"
  - name: "fraud-detection-model"
    team: "ml-platform"

change_policy:
  breaking_changes_notice_days: 14
  deprecation_notice_days: 30
  versioning_strategy: semantic

계약 강제 자동화 — CI/CD 통합

# contract_validator.py — GitHub Actions / GitLab CI에서 실행
import yaml
import pandas as pd
from dataclasses import dataclass
from typing import List

@dataclass
class ContractViolation:
    contract_name: str
    dimension: str
    field: str
    message: str
    severity: str  # "error" | "warning"

def validate_contract(df: pd.DataFrame, contract_path: str) -> List[ContractViolation]:
    violations = []

    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    schema_fields = {f["name"]: f for f in contract["schema"]["fields"]}

    for field_name, field_spec in schema_fields.items():

        if field_name not in df.columns:
            violations.append(ContractViolation(
                contract_name=contract["metadata"]["name"],
                dimension="schema",
                field=field_name,
                message=f"필드 '{field_name}'이 데이터에 없습니다",
                severity="error"
            ))
            continue

        if not field_spec.get("nullable", True):
            null_count = df[field_name].isnull().sum()
            if null_count > 0:
                violations.append(ContractViolation(
                    contract_name=contract["metadata"]["name"],
                    dimension="completeness",
                    field=field_name,
                    message=f"NOT NULL 위반: {null_count:,}개 NULL 발견",
                    severity="error"
                ))

        constraints = field_spec.get("constraints", {})
        if "allowed_values" in constraints:
            invalid = df[
                ~df[field_name].isin(constraints["allowed_values"])
                & df[field_name].notna()
            ]
            if len(invalid) > 0:
                violations.append(ContractViolation(
                    contract_name=contract["metadata"]["name"],
                    dimension="validity",
                    field=field_name,
                    message=f"허용값 위반: {invalid[field_name].unique()[:5].tolist()}",
                    severity="error"
                ))

        if "min" in constraints:
            below_min = df[df[field_name] < constraints["min"]]
            if len(below_min) > 0:
                violations.append(ContractViolation(
                    contract_name=contract["metadata"]["name"],
                    dimension="validity",
                    field=field_name,
                    message=f"최솟값({constraints['min']}) 미만: {len(below_min):,}건",
                    severity="error"
                ))

    return violations


if __name__ == "__main__":
    import sys

    df = pd.read_parquet("data/orders_sample.parquet")
    violations = validate_contract(df, "contracts/data_contract_orders.yaml")

    errors = [v for v in violations if v.severity == "error"]
    warnings = [v for v in violations if v.severity == "warning"]

    for v in violations:
        icon = "❌" if v.severity == "error" else "⚠️"
        print(f"{icon} [{v.dimension}] {v.field}: {v.message}")

    if errors:
        print(f"\n총 {len(errors)}개 오류 — 파이프라인을 중단합니다.")
        sys.exit(1)
    else:
        print(f"\n✅ 계약 검증 통과 (경고 {len(warnings)}개)")

계약 도구 생태계

도구특징오픈소스
dbt Contractsdbt 모델에 스키마 계약 선언, CI에서 Breaking Change 자동 감지
SchemataPython/TypeScript 계약 라이브러리, 스키마 진화 지원
Kafka Schema Registry스트리밍 데이터의 스키마 호환성 강제 (Confluent 제공)부분
OpenDataContractYAML 기반 오픈 표준 명세
Atlan계약 메타데이터를 카탈로그·계보·정책과 통합 관리

5. 데이터 카탈로그 구축과 운영

데이터 카탈로그란?

데이터 카탈로그는 조직의 모든 데이터 에셋에 대한 메타데이터를 중앙에서 관리하는 시스템이다. "이 데이터가 어디 있고, 무슨 의미인가, 신뢰할 수 있는가, 누가 책임지는가"에 답한다. 좋은 데이터 카탈로그는 데이터 디스커버리 시간을 수 시간에서 수 분으로 단축한다.

카탈로그 없는 환경:
  "이 숫자 어디서 나온 거죠?" → 슬랙 DM → 3명에게 물어봄 → 2시간 후 답변

카탈로그가 있는 환경:
  "이 숫자 어디서 나온 거죠?" → 카탈로그 검색 → 2분 만에 계보·오너·품질 확인

카탈로그에 담겨야 할 것들

메타데이터 유형포함 내용
기술적 메타데이터스키마, 데이터 타입, 파티션, 행 수, 마지막 갱신 시간, 저장 위치
비즈니스 메타데이터비즈니스 용어집(Glossary) 연결, 소유자, 설명, 사용 사례, 태그
운영 메타데이터파이프라인 실행 이력, 품질 점수, SLA 준수 여부, 인시던트 이력
거버넌스 메타데이터데이터 분류 (PII/공개/제한), 접근 정책, 보존 기간, 컴플라이언스 태그

2026년 주요 데이터 카탈로그 도구

도구특징적합 환경
AtlanAI 기반 메타데이터, 자연어 검색, 거버넌스 통합현대적 데이터 스택
Alation기업용, 행동 기반 추천, Soda와 통합대기업·금융
Collibra강력한 거버넌스 워크플로우, 규제 강한 산업금융·제약
OpenMetadata오픈소스, Self-hosted, 커스터마이즈 가능소규모 팀, 비용 민감
DataHub오픈소스, 대규모 확장성 (LinkedIn 개발)기술 기업
Databricks Unity CatalogDatabricks 생태계 내 통합 거버넌스Databricks 사용 조직

카탈로그 도입 로드맵

효과적인 카탈로그는 처음부터 모든 데이터를 등록하려 하지 않는다. 높은 비즈니스 영향도의 도메인부터 시작해 점진적으로 확장한다.

Phase 1 (1~4주): 기초 설정
  - 도구 선택 및 기술 스택 연결 (Snowflake, dbt, Airflow)
  - 가장 중요한 Gold 레이어 테이블 30개 등록
  - 데이터 오너 지정 및 비즈니스 설명 작성

Phase 2 (1~3개월): 자동화 확장
  - 메타데이터 자동 수집 크롤러 설정
  - dbt 문서를 카탈로그에 자동 동기화
  - PII 태깅 자동화
  - 데이터 품질 점수 카탈로그에 표시

Phase 3 (3~6개월): 거버넌스 심화
  - 접근 요청 워크플로우 카탈로그 통합
  - 비즈니스 용어집(Glossary) 구축
  - 계보 전체 시각화 활성화
  - 데이터 제품(Data Product) 단위 관리 전환

6. 데이터 계보 (Data Lineage)

데이터 계보란?

데이터 계보는 데이터가 소스에서 시작해 변환·적재·소비되는 전체 여정을 추적한다. "이 KPI 숫자가 어디서 왔는가?"에서 시작해 "이 소스 테이블이 변경되면 어떤 대시보드가 영향을 받는가?"까지 답한다.

계보의 3가지 레벨

① 테이블 레벨 계보: 어떤 테이블이 어떤 테이블에서 만들어지는가. fct_orders는 stg_orders와 dim_customers에서 만들어진다

② 컬럼 레벨 계보: 특정 컬럼이 어떤 소스 컬럼에서 파생됐는가. fct_orders.revenue는 stg_orders.amount_cents / 100에서 계산됨

③ 로우 레벨 계보: 특정 레코드의 소스까지 추적. 주로 감사·규제 목적. 가장 세밀하지만 구현 비용이 가장 크다.

OpenLineage — 계보 표준

OpenLineage는 데이터 계보를 수집하는 오픈 표준이다. Airflow, dbt, Spark, Flink 등 주요 도구가 OpenLineage 이벤트를 방출하도록 지원한다.

{
  "eventType": "COMPLETE",
  "eventTime": "2026-04-19T06:00:00Z",
  "run": {
    "runId": "3b452f3c-a462-4c78-bf8f-f9f553e5c8e1"
  },
  "job": {
    "namespace": "data-platform",
    "name": "dbt.fct_daily_revenue"
  },
  "inputs": [
    {
      "namespace": "snowflake://company.us-east-1",
      "name": "analytics.silver.stg_orders"
    },
    {
      "namespace": "snowflake://company.us-east-1",
      "name": "analytics.silver.stg_customers"
    }
  ],
  "outputs": [
    {
      "namespace": "snowflake://company.us-east-1",
      "name": "analytics.gold.fct_daily_revenue"
    }
  ]
}

영향 분석 (Impact Analysis) 활용

소스 테이블 스키마를 변경하기 전에 계보를 통해 다운스트림 영향을 자동으로 파악한다.


7. 데이터 옵저버빌리티

옵저버빌리티 vs 모니터링

모니터링옵저버빌리티
방식알려진 오류를 감시알 수 없는 오류를 발견
질문"이 파이프라인이 실행됐는가?""데이터가 신뢰할 수 있는가?"
접근임계값 기반 알림ML 기반 이상 탐지

데이터 옵저버빌리티는 소프트웨어 APM(Application Performance Monitoring)의 개념을 데이터 파이프라인에 적용한 것이다. 2026년 승리하는 기업들은 가장 많은 데이터를 가진 기업이 아니라, 가장 신뢰할 수 있는 데이터를 가진 기업이다.

"데이터 다운타임" 5대 신호

Monte Carlo가 정의한 데이터 다운타임 탐지 5대 차원:

신호의미
신선도 (Freshness)예상보다 오래된 데이터 ("어제 데이터가 왜 아직 안 들어왔지?")
볼륨 (Volume)예상보다 많거나 적은 행 수 (갑자기 0행 or 10배 급증)
스키마 (Schema)예상치 않은 컬럼 추가·삭제·타입 변경
분포 (Distribution)값의 통계적 특성 변화 (평균 금액이 갑자기 10배 증가)
계보 (Lineage)업스트림 에셋 변경이 다운스트림에 미치는 영향 자동 추적

Soda 실전 예시 — SQL 기반 품질 체크

Soda는 SQL 기반 품질 체크를 dbt·Airflow·CI/CD와 쉽게 통합할 수 있다.

# soda_checks_orders.yaml
# 실행: soda scan -d snowflake orders

checks for orders:

  # 신선도 검사
  - freshness(ordered_at) < 2h:
      name: "주문 데이터 신선도 2시간 이내"
      fail: when > 4h
      warn: when > 2h

  # 볼륨 이상 탐지
  - row_count > 0:
      name: "빈 테이블 감지"

  - row_count between 500 and 1000000:
      name: "일별 볼륨 정상 범위"
      warn: when not between 200 and 2000000

  # 완전성 검사
  - missing_count(order_id) = 0:
      name: "order_id NULL 없음"

  - missing_percent(customer_id) < 0.1%:
      name: "customer_id 완전성 99.9% 이상"

  # 유효성 검사
  - invalid_percent(order_status) < 0.01%:
      valid values: [placed, shipped, delivered, cancelled, refunded]
      name: "order_status 허용값 검증"

  # 고유성 검사
  - duplicate_count(order_id) = 0:
      name: "order_id 중복 없음"

  # 참조 무결성 검사
  - referential integrity (customer_id) must exist in customers (customer_id):
      name: "고객 ID 참조 무결성"

옵저버빌리티 도구 비교 2026

도구강점가격대오픈소스
Monte CarloML 기반 이상 탐지, 자동 계보, "데이터 다운타임" 개념 창시엔터프라이즈
SodaSQL 기반, dbt/Airflow 통합 우수, CI/CD 친화적무료 티어 존재부분
Great Expectations오픈소스, 풍부한 Expectation 라이브러리오픈소스
Metaplane자동 모니터링 설정, 웨어하우스 통합 간편SMB 친화
dbt 내장 테스트dbt 프로젝트 내 품질 검사, 추가 도구 불필요dbt 플랜 포함

8. PII 관리 & 컴플라이언스 자동화

PII 분류 체계

PII(Personally Identifiable Information, 개인식별정보)는 개인을 식별할 수 있는 모든 정보다. 2026년 EU AI Act 본격 시행, GDPR 강화, 국내 개인정보보호법 개정으로 데이터 파이프라인에서 PII 관리는 선택이 아닌 의무다.

분류예시 필드처리 방법
직접 식별자 (Direct ID)이름, 주민번호, 여권번호, 이메일익명화(Anonymization) 또는 토큰화(Tokenization)
간접 식별자 (Quasi-ID)생년월일, 주소, 우편번호, IP가명처리(Pseudonymization) 또는 마스킹(Masking)
민감 정보 (Sensitive)건강 정보, 금융 계좌, 신용 점수엄격한 접근 통제 + 암호화 + 감사 로그

데이터 마스킹 기법

# data_masking.py — 파이프라인 수집 시점에 적용

import hashlib
import re

class DataMasker:

    @staticmethod
    def tokenize(value: str, salt: str = "company_secret") -> str:
        """
        토큰화: 일관된 가명으로 변환 (조인 가능성 유지)
        customer_id 등 분석에서 JOIN이 필요한 필드에 적합
        """
        if value is None:
            return None
        return hashlib.sha256(f"{value}{salt}".encode()).hexdigest()[:16]

    @staticmethod
    def mask_email(email: str) -> str:
        """부분 마스킹: john.doe@company.com → jo***@company.com"""
        if not email or "@" not in email:
            return email
        local, domain = email.split("@", 1)
        masked_local = local[:2] + "***"
        return f"{masked_local}@{domain}"

    @staticmethod
    def mask_phone(phone: str) -> str:
        """전화번호 마스킹: 010-1234-5678 → 010-****-5678"""
        if not phone:
            return phone
        return re.sub(r'(\d{3})-(\d{4})-(\d{4})', r'\1-****-\3', phone)

    @staticmethod
    def mask_credit_card(cc: str) -> str:
        """신용카드 마스킹: 4111-1111-1111-1234 → ****-****-****-1234"""
        if not cc:
            return cc
        digits = re.sub(r'\D', '', cc)
        return f"****-****-****-{digits[-4:]}"

    @staticmethod
    def anonymize(value) -> str:
        """완전 익명화: 분석에서 개인 식별이 전혀 필요 없는 경우"""
        return "REDACTED"


def apply_pii_masking(df):
    """Bronze → Silver 전환 시 PII 마스킹 적용"""
    masker = DataMasker()

    if "customer_email" in df.columns:
        df["customer_email"] = df["customer_email"].apply(masker.mask_email)

    if "phone_number" in df.columns:
        df["phone_number"] = df["phone_number"].apply(masker.mask_phone)

    if "customer_id" in df.columns:
        df["customer_id"] = df["customer_id"].apply(masker.tokenize)

    if "ssn" in df.columns:
        df.drop(columns=["ssn"], inplace=True)

    return df

Medallion 아키텍처에서의 PII 흐름


9. 접근 제어 & 역할 설계

최소 권한 원칙 (Principle of Least Privilege)

모든 사용자와 서비스는 업무에 필요한 최소한의 권한만 가져야 한다. 데이터 플랫폼에서 가장 자주 위반되는 보안 원칙이다.

-- Snowflake RBAC 예시

-- 역할 계층 설계
CREATE ROLE analyst_gold;      -- Gold 읽기 전용 (전체 분석팀)
CREATE ROLE analyst_silver;    -- Silver 읽기 전용
CREATE ROLE analyst_bronze;    -- Bronze 읽기 전용
CREATE ROLE data_engineer;     -- Silver/Gold 쓰기 + Bronze 읽기
CREATE ROLE pii_admin;         -- Bronze PII 컬럼 포함 전체 읽기

-- 역할 상속
GRANT ROLE analyst_bronze TO ROLE analyst_silver;
GRANT ROLE analyst_silver TO ROLE analyst_gold;
GRANT ROLE analyst_gold TO ROLE data_engineer;

-- 테이블 수준 권한
GRANT SELECT ON ALL TABLES IN SCHEMA gold TO ROLE analyst_gold;
GRANT SELECT ON ALL TABLES IN SCHEMA silver TO ROLE analyst_silver;

-- 컬럼 수준 동적 마스킹 (Snowflake Dynamic Data Masking)
CREATE MASKING POLICY email_mask AS (val STRING) RETURNS STRING ->
  CASE
    WHEN CURRENT_ROLE() IN ('pii_admin') THEN val
    WHEN CURRENT_ROLE() IN ('data_engineer') THEN
      REGEXP_REPLACE(val, '(^.{2}).*(@.*)', '\\1***\\2')
    ELSE '***REDACTED***'
  END;

ALTER TABLE silver.customers
MODIFY COLUMN email
SET MASKING POLICY email_mask;

데이터 접근 요청 워크플로우


10. 실전 체크리스트

데이터 품질

  • 모든 Gold 테이블에 6대 품질 차원 검사가 적용돼 있는가?
  • 파이프라인 각 단계(Bronze→Silver, Silver→Gold)에 품질 게이트가 존재하는가?
  • 데이터셋별 품질 SLA 등급(Gold/Silver/Bronze)이 정의돼 있는가?
  • 품질 검사 실패 시 자동 알림과 파이프라인 차단이 동작하는가?
  • dbt tests 또는 Great Expectations / Soda가 CI/CD에 통합돼 있는가?

거버넌스 프레임워크

  • 모든 주요 데이터셋에 데이터 오너가 지정돼 있는가?
  • 데이터 분류 체계(공개/내부/제한/기밀)가 정의·적용돼 있는가?
  • 거버넌스 정책이 문서가 아닌 코드(정책-as-코드)로 관리되는가?
  • 거버넌스 프레임워크 유형(중앙/분산/하이브리드)이 조직 구조에 맞게 선택됐는가?

데이터 계약

  • 주요 Gold 테이블에 데이터 계약(YAML 명세)이 작성돼 있는가?
  • 계약이 CI/CD 파이프라인에서 자동 검증되는가?
  • Breaking Change 시 소비자에게 사전 통보(14일 이상) 프로세스가 있는가?
  • Kafka 스트리밍 데이터에 Schema Registry가 적용돼 있는가?

데이터 카탈로그 & 계보

  • 데이터 카탈로그가 도입돼 있으며 주요 에셋이 등록돼 있는가?
  • 데이터 계보(소스 → 변환 → 소비)가 자동으로 추적되는가?
  • 스키마 변경 전 다운스트림 영향 분석이 가능한가?
  • 비즈니스 용어집(Glossary)이 카탈로그에 연결돼 있는가?

PII & 컴플라이언스

  • PII 컬럼이 자동으로 탐지·태깅되는가?
  • Bronze→Silver 전환 시 PII 마스킹이 자동 적용되는가?
  • 컬럼 레벨 접근 제어(Dynamic Data Masking)가 적용돼 있는가?
  • 데이터 접근 요청/승인/만료 워크플로우가 자동화돼 있는가?
  • 데이터 보존 기간 정책이 정의·자동 적용되는가?
  • GDPR/CCPA/개인정보보호법 요건을 반영한 감사 로그가 생성되는가?

마치며

데이터 품질과 거버넌스는 "한 번 구축하면 끝나는" 프로젝트가 아니다. 비즈니스가 변하고, 데이터가 늘어나고, 규제가 강화되는 만큼 지속적으로 진화하는 살아있는 시스템이어야 한다.

2026년 가장 중요한 패러다임 전환은 이것이다: 거버넌스를 "데이터 엔지니어링을 방해하는 규제"로 보는 대신, "데이터 제품의 신뢰성을 보장하는 엔지니어링 실천" 으로 바라보는 것. 거버넌스가 내재화된 파이프라인은 더 빠른 의사 결정, 더 신뢰할 수 있는 AI, 그리고 컴플라이언스 위반 걱정 없는 운영을 가능하게 한다.

다음 파트에서는 이 모든 것을 뒷받침하는 클라우드 인프라 설계와 비용 최적화(FinOps) — IaC, 멀티클라우드 전략, 컴퓨트 비용 거버넌스 — 를 심층 분석한다.


Part 5 예고: 클라우드 & 인프라 심층 가이드

  • AWS vs GCP vs Azure 데이터 플랫폼 비교
  • Terraform으로 데이터 인프라 코드화 (IaC)
  • FinOps — 클라우드 데이터 비용 최적화 실전
  • Kubernetes 기반 데이터 플랫폼 운영
  • 멀티클라우드 & 하이브리드 클라우드 전략

이 글 공유하기

시리즈 내비게이션

Data Engineering 플레이북

4 / 7 · 4

같은 주제 더 보기·대표 시리즈로 시작

English

최신 글을 RSS로 받아보세요

뉴스레터 오픈 전에는 RSS로 먼저 업데이트를 받아보실 수 있습니다.

RSS 구독 안내 보기