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 & 팀 운영 플레이북 (연재 예정)
목차
- 왜 데이터 품질과 거버넌스인가?
- 데이터 품질 6대 차원 프레임워크
- DataGovOps — 거버넌스를 코드로
- 데이터 계약 (Data Contract) — 생산자와 소비자 사이의 약속
- 데이터 카탈로그 구축과 운영
- 데이터 계보 (Data Lineage) — 데이터의 여정 추적
- 데이터 옵저버빌리티 — 신뢰의 인프라
- PII 관리 & 컴플라이언스 자동화
- 접근 제어 & 역할 설계
- 실전 체크리스트
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 | 모니터링 주기 |
|---|---|---|---|---|
| Gold | KPI, 재무 보고, 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 Contracts | dbt 모델에 스키마 계약 선언, CI에서 Breaking Change 자동 감지 | ✅ |
| Schemata | Python/TypeScript 계약 라이브러리, 스키마 진화 지원 | ✅ |
| Kafka Schema Registry | 스트리밍 데이터의 스키마 호환성 강제 (Confluent 제공) | 부분 |
| OpenDataContract | YAML 기반 오픈 표준 명세 | ✅ |
| Atlan | 계약 메타데이터를 카탈로그·계보·정책과 통합 관리 | ❌ |
5. 데이터 카탈로그 구축과 운영
데이터 카탈로그란?
데이터 카탈로그는 조직의 모든 데이터 에셋에 대한 메타데이터를 중앙에서 관리하는 시스템이다. "이 데이터가 어디 있고, 무슨 의미인가, 신뢰할 수 있는가, 누가 책임지는가"에 답한다. 좋은 데이터 카탈로그는 데이터 디스커버리 시간을 수 시간에서 수 분으로 단축한다.
카탈로그 없는 환경:
"이 숫자 어디서 나온 거죠?" → 슬랙 DM → 3명에게 물어봄 → 2시간 후 답변
카탈로그가 있는 환경:
"이 숫자 어디서 나온 거죠?" → 카탈로그 검색 → 2분 만에 계보·오너·품질 확인
카탈로그에 담겨야 할 것들
| 메타데이터 유형 | 포함 내용 |
|---|---|
| 기술적 메타데이터 | 스키마, 데이터 타입, 파티션, 행 수, 마지막 갱신 시간, 저장 위치 |
| 비즈니스 메타데이터 | 비즈니스 용어집(Glossary) 연결, 소유자, 설명, 사용 사례, 태그 |
| 운영 메타데이터 | 파이프라인 실행 이력, 품질 점수, SLA 준수 여부, 인시던트 이력 |
| 거버넌스 메타데이터 | 데이터 분류 (PII/공개/제한), 접근 정책, 보존 기간, 컴플라이언스 태그 |
2026년 주요 데이터 카탈로그 도구
| 도구 | 특징 | 적합 환경 |
|---|---|---|
| Atlan | AI 기반 메타데이터, 자연어 검색, 거버넌스 통합 | 현대적 데이터 스택 |
| Alation | 기업용, 행동 기반 추천, Soda와 통합 | 대기업·금융 |
| Collibra | 강력한 거버넌스 워크플로우, 규제 강한 산업 | 금융·제약 |
| OpenMetadata | 오픈소스, Self-hosted, 커스터마이즈 가능 | 소규모 팀, 비용 민감 |
| DataHub | 오픈소스, 대규모 확장성 (LinkedIn 개발) | 기술 기업 |
| Databricks Unity Catalog | Databricks 생태계 내 통합 거버넌스 | 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 Carlo | ML 기반 이상 탐지, 자동 계보, "데이터 다운타임" 개념 창시 | 엔터프라이즈 | ❌ |
| Soda | SQL 기반, 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 기반 데이터 플랫폼 운영
- 멀티클라우드 & 하이브리드 클라우드 전략