OpenLineage 표준

# 사용자 정의 패싯 스키마 정의 (JSON Schema)
{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "_producer": {
      "type": "string"
    },
    "_schemaURL": {
      "type": "string",
      "format": "uri"
    },
    "costCenter": {
      "type": "string",
      "description": "데이터 처리 비용이 청구될 비용 센터"
    },
    "dataClassification": {
      "type": "string",
      "enum": ["public", "internal", "confidential", "restricted"]
    },
    "retentionPolicy": {
      "type": "object",
      "properties": {
        "days": { "type": "integer" },
        "reason": { "type": "string" }
      }
    },
    "slaRequirements": {
      "type": "object",
      "properties": {
        "maxLatencyMinutes": { "type": "integer" },
        "availability": { "type": "number" },
        "freshnessHours": { "type": "integer" }
      }
    }
  }
}

# Python에서 사용자 정의 패싯 사용
from openlineage.client.run import Dataset, Run, Job
from openlineage.client.facet import DatasetFacet

# 사용자 정의 패싯 클래스
class CustomGovernanceFacet(DatasetFacet):
    costCenter: str
    dataClassification: str
    retentionPolicy: dict
    slaRequirements: dict

# OpenLineage 이벤트에 추가
dataset = Dataset(
    namespace="production",
    name="warehouse.sensitive_data",
    facets={
        "customGovernance": CustomGovernanceFacet(
            costCenter="DATA-PLATFORM",
            dataClassification="restricted",
            retentionPolicy={
                "days": 2555,  # 7 years
                "reason": "GDPR Article 17"
            },
            slaRequirements={
                "maxLatencyMinutes": 30,
                "availability": 99.9,
                "freshnessHours": 2
            }
        )
    }
)

# dbt 프로젝트에서 OpenLineage 설정

# 1. dbt-ol 패키지 설치
pip install dbt-ol

# 2. dbt profiles.yml 구성
my_project:
  outputs:
    prod:
      type: snowflake
      account: my_account
      # ... 기타 연결 설정

      # OpenLineage 구성 추가
      openlineage:
        transport:
          type: http
          url: http://marquez:5000
          endpoint: /api/v1/lineage
        namespace: production
        job_name: dbt_daily_transform

# 3. dbt 실행 시 자동으로 계보 이벤트 전송
dbt run --models +customers

# 생성되는 OpenLineage 이벤트:
# - 각 모델에 대한 RUN 이벤트
# - 소스 테이블 READ 이벤트
# - 변환된 테이블 WRITE 이벤트
# - SQL 변환 로직 (패싯)
# - 컬럼 레벨 계보 (패싯)

# docker-compose.yml
version: '3.8'

services:
  # PostgreSQL 데이터베이스
  postgres:
    image: postgres:14
    environment:
      POSTGRES_USER: marquez
      POSTGRES_PASSWORD: marquez
      POSTGRES_DB: marquez
    volumes:
      - postgres-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U marquez"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Marquez API 서버
  marquez:
    image: marquezproject/marquez:latest
    environment:
      MARQUEZ_PORT: 5000
      MARQUEZ_DB_HOST: postgres
      MARQUEZ_DB_PORT: 5432
      MARQUEZ_DB_NAME: marquez
      MARQUEZ_DB_USER: marquez
      MARQUEZ_DB_PASSWORD: marquez
    ports:
      - "5000:5000"
      - "5001:5001"  # Admin API
    depends_on:
      postgres:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:5000/api/v1/namespaces"]
      interval: 30s
      timeout: 10s
      retries: 3

  # Marquez Web UI
  marquez-web:
    image: marquezproject/marquez-web:latest
    environment:
      MARQUEZ_HOST: marquez
      MARQUEZ_PORT: 5000
    ports:
      - "3000:3000"
    depends_on:
      - marquez

volumes:
  postgres-data:

# 배포
docker-compose up -d

# 상태 확인
curl http://localhost:5000/api/v1/namespaces

# Web UI 접속
# http://localhost:3000

# 1. 네임스페이스 조회
GET http://localhost:5000/api/v1/namespaces

# 2. 특정 데이터셋의 계보 가져오기
GET http://localhost:5000/api/v1/lineage?
    nodeId=dataset:production:warehouse.customers

# 3. 작업 실행 이력 조회
GET http://localhost:5000/api/v1/namespaces/production/jobs/etl_customer_daily/runs

# 4. 데이터셋 메타데이터 조회
GET http://localhost:5000/api/v1/namespaces/production/datasets/warehouse.customers

# 5. 검색
GET http://localhost:5000/api/v1/search?q=customer&filter=DATASET

# Python 클라이언트 사용
from marquez_client import MarquezClient

client = MarquezClient(url="http://localhost:5000")

# 계보 조회
lineage = client.get_lineage(
    node_id="dataset:production:warehouse.customers",
    depth=5
)

# 그래프 탐색
for node in lineage.graph:
    print(f"{node.type}: {node.data.name}")
    print(f"  업스트림: {len(node.in_edges)}")
    print(f"  다운스트림: {len(node.out_edges)}")

# OpenLineage 이벤트를 Egeria로 전송

from pyegeria import EgeriaClient
from openlineage.client import OpenLineageClient

class EgeriaLineageBridge:
    """OpenLineage 이벤트를 Egeria 메타데이터로 변환"""

    def __init__(self, egeria_url, ol_client):
        self.egeria = EgeriaClient(egeria_url)
        self.ol_client = ol_client

    def sync_lineage_to_egeria(self, ol_event):
        """OpenLineage 이벤트를 Egeria 엔티티로 변환"""

        # 1. 데이터셋 → Egeria DataSet 엔티티
        for input_dataset in ol_event.inputs:
            self.create_or_update_dataset(input_dataset)

        for output_dataset in ol_event.outputs:
            self.create_or_update_dataset(output_dataset)

        # 2. 작업 → Egeria Process 엔티티
        process_guid = self.create_or_update_process(ol_event.job, ol_event.run)

        # 3. 계보 관계 생성
        for input_ds in ol_event.inputs:
            self.create_lineage_relationship(
                source=input_ds.name,
                target=process_guid,
                relationship_type="DataFlow"
            )

        for output_ds in ol_event.outputs:
            self.create_lineage_relationship(
                source=process_guid,
                target=output_ds.name,
                relationship_type="DataFlow"
            )

    def create_or_update_dataset(self, dataset):
        """Egeria에 데이터셋 등록"""
        entity = {
            "typeName": "DataSet",
            "attributes": {
                "qualifiedName": f"{dataset.namespace}/{dataset.name}",
                "displayName": dataset.name,
                "description": dataset.facets.get("documentation", {}).get("description")
            }
        }

        # 스키마 정보 추가
        if "schema" in dataset.facets:
            entity["schemaType"] = self.convert_schema_facet(
                dataset.facets["schema"]
            )

        return self.egeria.create_entity(entity)

    def create_lineage_relationship(self, source, target, relationship_type):
        """계보 관계 생성"""
        return self.egeria.create_relationship(
            relationship_type,
            source_guid=source,
            target_guid=target
        )

# Marquez Prometheus 메트릭 엔드포인트
GET http://localhost:5001/metrics

# 주요 메트릭:
# - marquez_events_total{type="RUN"}
# - marquez_events_latency_seconds
# - marquez_api_requests_total{endpoint="/lineage"}
# - marquez_db_connections_active
# - marquez_storage_bytes_total

# Grafana 대시보드 쿼리 예시
# 이벤트 수집 속도
rate(marquez_events_total[5m])

# p95 지연시간
histogram_quantile(0.95, marquez_events_latency_seconds_bucket)

# API 오류율
rate(marquez_api_requests_total{status=~"5.."}[5m])
/ rate(marquez_api_requests_total[5m])

사례: 전자상거래 플랫폼의 일일 매출 보고서

1. 원본 데이터 (Kafka)
   - 주제: user_events, transactions
   - OpenLineage 이벤트: Kafka Connect로 자동 생성
   - 패싯: 스키마, 파티션, 오프셋

2. 스트림 처리 (Spark Streaming)
   - 입력: Kafka 주제
   - 처리: 필터링, 집계, 조인
   - 출력: S3에 Parquet
   - OpenLineage 이벤트: SparkListener로 자동 생성
   - 패싯: Spark 설정, 처리 메트릭

3. 배치 변환 (dbt)
   - 입력: S3 Parquet 파일
   - 변환: SQL 모델 (고객 세그먼트, 제품 분석)
   - 출력: Snowflake 테이블
   - OpenLineage 이벤트: dbt-ol로 자동 생성
   - 패싯: SQL 로직, 컬럼 계보

4. 데이터 품질 (Great Expectations)
   - 검증: 행 수, NULL 검사, 범위 검사
   - OpenLineage 이벤트: GE 통합으로 생성
   - 패싯: 검증 결과, 품질 메트릭

5. 오케스트레이션 (Airflow)
   - DAG: daily_revenue_report
   - 작업: extract → transform → validate → load
   - OpenLineage 이벤트: Airflow 플러그인으로 생성
   - 패싯: 작업 의존성, 스케줄, 소유자

6. BI 도구 (Tableau)
   - 데이터 소스: Snowflake warehouse.daily_revenue
   - OpenLineage 이벤트: 사용자 정의 커넥터
   - 패싯: 대시보드 메타데이터, 사용 통계

최종 결과:
- 완전한 엔드투엔드 계보: Kafka → Spark → S3 → dbt → Snowflake → Tableau
- 각 단계의 메타데이터 캡처
- 자동화된 품질 추적
- 영향 분석 가능
- 규정 준수 감사 추적