Overwriting 패턴#

1
# === TRUNCATE 기반 워크플로우 ===
2
def truncate_based_idempotency(db, execution_date, data):
3
    table_name = f"visits_week_{execution_date.isocalendar()[1]}"
4

5
    # 1. 테이블 없으면 생성
6
    db.execute(f"CREATE TABLE IF NOT EXISTS {table_name} (LIKE visits_template)")
7

8
    # 2. 뷰에 포함
9
    db.execute(f"CREATE OR REPLACE VIEW visits AS SELECT * FROM {table_name} UNION ALL ...")
10

11
    # 3. TRUNCATE — 메타데이터 연산, 매우 빠름
12
    db.execute(f"TRUNCATE TABLE {table_name}")
13

14
    # 4. 데이터 적재
15
    db.execute(f"INSERT INTO {table_name} SELECT * FROM {data}")
16

17

18
# === DROP 기반 워크플로우 ===
19
def drop_based_idempotency(db, execution_date, data):
20
    table_name = f"visits_week_{execution_date.isocalendar()[1]}"
21

22
    # 1. 뷰에서 먼저 제거 (소비자 에러 방지)
23
    db.execute(f"-- remove {table_name} from view")
24

25
    # 2. DROP — 테이블 자체를 삭제
26
    db.execute(f"DROP TABLE IF EXISTS {table_name}")
27

28
    # 3. 새로 생성
29
    db.execute(f"CREATE TABLE {table_name} (LIKE visits_template)")
30

31
    # 4. 뷰에 다시 포함
32
    db.execute(f"CREATE OR REPLACE VIEW visits AS SELECT * FROM {table_name} UNION ALL ...")
33

34
    # 5. 데이터 적재
35
    db.execute(f"INSERT INTO {table_name} SELECT * FROM {data}")

1
# Full Load용 단순화된 Fast Metadata Cleaner
2
def full_load_fast_clean(db, data):
3
    # DROP → CREATE → LOAD (매 실행마다)
4
    db.execute("DROP TABLE IF EXISTS devices")
5
    db.execute("CREATE TABLE devices (type TEXT, full_name TEXT, version TEXT)")
6
    db.execute(f"COPY devices FROM '{data}'")

1
# Apache Spark — overwrite 모드
2
input_data = spark.read.json("s3://input/visits/date=2024-01-15")
3

4
# mode('overwrite')가 기존 데이터를 삭제 후 새로 쓴다
5
input_data.write.mode('overwrite').parquet("s3://output/visits/date=2024-01-15")

1
# 전통적인 SQL 방식
2
def overwrite_with_delete_insert(db, target_date, staging_table):
3
    # 1단계: 해당 날짜 데이터 삭제
4
    db.execute(f"DELETE FROM visits WHERE event_date = '{target_date}'")
5

6
    # 2단계: 새 데이터 삽입
7
    db.execute(f"""
8
        INSERT INTO visits
9
        SELECT * FROM {staging_table}
10
        WHERE event_date = '{target_date}'
11
    """)

1
# Databricks, Snowflake 등에서 지원
2
# DELETE + INSERT를 하나의 명령으로 수행
3
db.execute("""
4
    INSERT OVERWRITE INTO devices
5
    SELECT * FROM devices_staging
6
""")

1
# Delta Lake — 특정 파티션만 덮어쓰기
2
(input_data.write
3
    .format("delta")
4
    .mode("overwrite")
5
    .option("replaceWhere", "event_date = '2024-01-15'")
6
    .save("s3://output/visits"))

1
# 덮어쓰기 후에도 이전 버전 조회 가능 (Delta Lake)
2
spark.read.format("delta").option("versionAsOf", 3).load("s3://output/visits")
3

4
# 오래된 파일을 실제로 삭제하려면 VACUUM 실행
5
from delta.tables import DeltaTable
6
dt = DeltaTable.forPath(spark, "s3://output/visits")
7
dt.vacuum(retentionHours=168)  # 7일 이전 파일 삭제