데이터 출처와 추적

from datetime import datetime
import requests

def emit_provenance_event(event_type, dataset, metadata):
    """데이터 처리 중 출처 이벤트 전송"""
    event = {
        "eventType": event_type,
        "timestamp": datetime.utcnow().isoformat(),
        "dataset": {
            "namespace": dataset["namespace"],
            "name": dataset["name"]
        },
        "metadata": metadata,
        "producer": "etl_pipeline_v2.3"
    }

    # 출처 수집기로 전송
    response = requests.post(
        "http://lineage-collector:8080/api/v1/events",
        json=event
    )
    return response.status_code == 200

# 사용 예시
emit_provenance_event(
    event_type="READ",
    dataset={"namespace": "warehouse", "name": "customers"},
    metadata={
        "rowsRead": 45892,
        "bytesRead": 12458976,
        "partitions": ["2025-01-26"]
    }
)

import sqlparse
from sqlalchemy import event, create_engine

class LineageTracker:
    """SQL 쿼리에서 계보 자동 추출"""

    def __init__(self, lineage_api):
        self.lineage_api = lineage_api

    def track_query(self, statement, parameters):
        """쿼리 실행 전후 계보 추적"""
        parsed = sqlparse.parse(statement)[0]

        # 테이블 참조 추출
        tables_read = self._extract_source_tables(parsed)
        tables_written = self._extract_target_tables(parsed)

        # 출처 이벤트 생성
        self.lineage_api.record_event({
            "query": statement,
            "sources": tables_read,
            "targets": tables_written,
            "timestamp": datetime.utcnow(),
            "user": self._get_current_user()
        })

    def _extract_source_tables(self, parsed_query):
        """FROM 및 JOIN 절에서 소스 테이블 추출"""
        # SQL AST 파싱 로직
        tables = []
        for token in parsed_query.tokens:
            if token.ttype is sqlparse.tokens.Keyword and token.value.upper() == 'FROM':
                # 테이블 추출
                pass
        return tables

# SQLAlchemy에 연결
engine = create_engine('postgresql://user:pass@host/db')
tracker = LineageTracker(lineage_api_client)

@event.listens_for(engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    tracker.track_query(statement, parameters)

from pyspark.sql import SparkSession
from openlineage.spark import SparkLineageListener

# OpenLineage 리스너와 함께 Spark 세션 구성
spark = SparkSession.builder \
    .appName("LineageTrackedApp") \
    .config("spark.extraListeners", "io.openlineage.spark.agent.OpenLineageSparkListener") \
    .config("spark.openlineage.transport.type", "http") \
    .config("spark.openlineage.transport.url", "http://marquez:5000") \
    .getOrCreate()

# 이제 모든 Spark 작업이 자동으로 추적됨
df = spark.read.parquet("s3://data-lake/raw/orders/")
df_transformed = df.filter(df.status == "completed") \
                   .groupBy("customer_id") \
                   .count()
df_transformed.write.parquet("s3://data-lake/processed/customer_orders/")

# 출처가 자동으로 기록됨:
# READ: s3://data-lake/raw/orders/
# WRITE: s3://data-lake/processed/customer_orders/
# TRANSFORMATION: filter + groupBy + count

# 1. 데이터셋의 전체 업스트림 계보 가져오기
GET /api/v1/provenance/upstream?dataset=warehouse.customer_summary&depth=5

# 2. 특정 컬럼의 출처 추적
GET /api/v1/provenance/column?dataset=warehouse.customers&column=email

# 3. 시간 범위 내 모든 변환 찾기
GET /api/v1/provenance/activities?start=2025-01-20&end=2025-01-26&dataset=sales

# 4. 특정 사용자가 생성한 데이터셋 찾기
GET /api/v1/provenance/entities?attributedTo=john.doe@company.com

# 5. 실패한 실행의 영향 평가
GET /api/v1/provenance/impact?execution=exec_20250126_143022&status=failed

// 특정 PII 필드를 사용하는 모든 다운스트림 리포트 찾기
MATCH path = (source:Column {name: 'customer_email', pii: true})-[:FLOWS_TO*]->(target:Report)
WHERE NOT (target)-[:FLOWS_TO]->()
RETURN target.name, length(path) as depth
ORDER BY depth DESC

// 지난 7일간 3회 이상 실패한 모든 파이프라인 찾기
MATCH (activity:ETL)-[:EXECUTION]->(run:Run)
WHERE run.status = 'FAILED'
  AND run.timestamp > datetime() - duration('P7D')
WITH activity, count(run) as failures
WHERE failures >= 3
RETURN activity.name, failures
ORDER BY failures DESC

from kafka import KafkaProducer
import json

class StreamingLineageTracker:
    """실시간 출처 이벤트를 Kafka로 전송"""

    def __init__(self, bootstrap_servers):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.topic = 'lineage.events'

    def track_stream_processing(self, stream_name, input_topics, output_topic, transform):
        """스트림 처리 출처 추적"""
        event = {
            "eventTime": datetime.utcnow().isoformat(),
            "stream": stream_name,
            "inputs": [{"topic": t, "partitions": "all"} for t in input_topics],
            "output": {"topic": output_topic},
            "transformation": transform,
            "processingMode": "STREAMING"
        }

        self.producer.send(self.topic, value=event)
        self.producer.flush()

# 사용 예시
tracker = StreamingLineageTracker(['kafka-broker:9092'])

# Flink/Spark Streaming 작업 추적
tracker.track_stream_processing(
    stream_name="customer_events_aggregator",
    input_topics=["user_clicks", "user_purchases"],
    output_topic="customer_360",
    transform="sessionization + enrichment + aggregation"
)