5. Immutable Dataset && Proxy#

1
from datetime import datetime
2

3
def proxy_view_based(db, execution_time: datetime, data):
4
    # 실행 시점으로 고유한 내부 테이블명 생성
5
    table_suffix = execution_time.strftime('%Y%m%d_%H%M%S')
6
    internal_table = f"devices_internal_{table_suffix}"
7

8
    # 1. 새 테이블에 데이터 적재 (한 번만 쓰기)
9
    db.execute(f"CREATE TABLE {internal_table} (type TEXT, full_name TEXT, version TEXT)")
10
    db.execute(f"COPY {internal_table} FROM '{data}'")
11

12
    # 2. 뷰를 최신 테이블로 갱신
13
    db.execute(f"CREATE OR REPLACE VIEW devices AS SELECT * FROM {internal_table}")
14

15
    # 3. 이전 테이블의 쓰기 권한 제거 (불변성 강제)
16
    db.execute(f"REVOKE INSERT, UPDATE, DELETE ON {internal_table} FROM pipeline_user")
17

18

19
# 재실행 시: 다른 timestamp → 다른 테이블명 → 이전 데이터 보존
20
proxy_view_based(db, datetime(2024, 1, 15, 10, 0), "/data/v1.csv")
21
# → devices_internal_20240115_100000
22

23
proxy_view_based(db, datetime(2024, 1, 15, 10, 5), "/data/v1.csv")
24
# → devices_internal_20240115_100500 (별도 테이블, 이전 것도 남아있음)

1
import json
2

3
def proxy_manifest_based(storage, execution_time: datetime, data_path: str):
4
    # 1. 새 경로에 데이터 쓰기
5
    versioned_path = f"s3://lake/devices/{execution_time.isoformat()}/"
6
    storage.copy(data_path, versioned_path)
7

8
    # 2. 매니페스트 파일 갱신 — 최신 경로만 가리킴
9
    manifest = {"current_version": versioned_path, "updated_at": execution_time.isoformat()}
10
    storage.write("s3://lake/devices/_manifest.json", json.dumps(manifest))
11

12
    # 3. 이전 버전 파일에 WORM 잠금 적용
13
    previous_versions = storage.list("s3://lake/devices/", exclude="_manifest.json")
14
    for version_path in previous_versions:
15
        if version_path != versioned_path:
16
            storage.set_object_lock(version_path)  # Write Once Read Many
17

18

19
# 소비자는 매니페스트를 읽어서 최신 경로를 알아냄
20
def read_latest(storage):
21
    manifest = json.loads(storage.read("s3://lake/devices/_manifest.json"))
22
    return storage.read(manifest["current_version"])

1
def proxy_versioned(spark, data):
2
    # 그냥 overwrite — 이전 버전은 테이블 포맷이 자동 보존
3
    data.write.format("delta").mode("overwrite").save("s3://lake/devices")
4

5
    # 과거 버전 조회도 가능
6
    # spark.read.format("delta").option("versionAsOf", 3).load("s3://lake/devices")

1
from airflow.operators.python import get_current_context
2

3
def get_devices_table_name() -> str:
4
    """파이프라인 시작 시간으로 고유 테이블명 생성"""
5
    context = get_current_context()
6
    dag_run = context['dag_run']
7
    table_suffix = dag_run.start_date.strftime('%Y%m%d_%H%M%S_%f')
8
    return f"dedp.devices_internal_{table_suffix}"
9

10
# 같은 파이프라인 인스턴스 재실행 → 다른 start_date → 다른 테이블
11
# → 이전 테이블은 보존됨 → 불변성 유지