1
# 좋은 키 생성: 불변 속성 사용
2
def generate_session_key_immutable(user_id: int, first_append_time: int) -> str:
3
    """append_time은 브로커가 기록한 시간 — 재시도해도 변하지 않음"""
4
    return f"{user_id}_{first_append_time}"
5

6
# 첫 실행
7
key = generate_session_key_immutable(user_id=1, first_append_time=957)
8
print(key)  # "1_957"
9

10
# 장애 후 재시도 — 같은 키 생성
11
key = generate_session_key_immutable(user_id=1, first_append_time=957)
12
print(key)  # "1_957" → 덮어쓰기 → 중복 없음

1
# 위험한 키 생성: event_time 사용
2
def generate_session_key_mutable(user_id: int, events: list) -> str:
3
    """event_time 기반 — late data가 올 경우 키가 변함"""
4
    min_event_time = min(e['event_time'] for e in events)
5
    return f"{user_id}_{min_event_time}"
6

7
# 첫 실행: 09:56, 09:57 이벤트
8
events_run1 = [
9
    {'event_time': '09:56', 'append_time': '09:57'},
10
    {'event_time': '09:57', 'append_time': '09:58'},
11
]
12
key1 = generate_session_key_mutable(1, events_run1)
13
print(key1)  # "1_09:56"
14

15
# 장애 후 재시도: late data(09:55)가 추가됨
16
events_run2 = [
17
    {'event_time': '09:55', 'append_time': '09:59'},  # ← late data!
18
    {'event_time': '09:56', 'append_time': '09:57'},
19
    {'event_time': '09:57', 'append_time': '09:58'},
20
]
21
key2 = generate_session_key_mutable(1, events_run2)
22
print(key2)  # "1_09:55" ← 키가 바뀜! 이전 "1_09:56"은 유령 데이터로 남음

1
from datetime import date
2

3
# 배치 잡: 실행 날짜로 파일명 생성 → 재실행해도 같은 파일 덮어쓰기
4
def get_output_path(execution_date: date) -> str:
5
    return f"s3://output/sessions/{execution_date.isoformat()}.parquet"
6

7
# 2024-11-20 실행 → "s3://output/sessions/2024-11-20.parquet"
8
# 2024-11-20 재실행 → 같은 경로 → 덮어쓰기 → 멱등

1
-- 각 유저의 첫 번째 ingestion_time을 기준으로 세션 키 생성
2
SELECT
3
    user_id,
4
    FIRST_VALUE(ingestion_time) OVER (
5
        PARTITION BY user_id
6
        ORDER BY ingestion_time ASC
7
    ) AS first_append_time
8
FROM visits

1
# 트랜잭션 패턴의 기본 구조
2
def transactional_write(db, data_batches: list):
3
    try:
4
        # STEP 1: 트랜잭션 시작
5
        db.execute("BEGIN")
6

7
        # STEP 2: 데이터 쓰기 (아직 소비자에게 보이지 않음)
8
        for batch in data_batches:
9
            db.execute(f"INSERT INTO visits SELECT * FROM {batch}")
10

11
        # STEP 3a: 성공 → commit (이 시점에 모든 데이터가 한 번에 공개)
12
        db.execute("COMMIT")
13

14
    except Exception:
15
        # STEP 3b: 실패 → rollback (아무것도 남지 않음)
16
        db.execute("ROLLBACK")
17
        raise

1
# === 단일 프로세스 (DW에서 SQL 실행) ===
2
# 트랜잭션이 선언적이고 DB가 완전히 관리
3
def elt_transactional(db):
4
    db.execute("BEGIN")
5
    db.execute("MERGE INTO devices USING staging_1 ON ...")
6
    db.execute("MERGE INTO devices USING staging_2 ON ...")
7
    db.execute("COMMIT")
8
    # staging_2에 오류가 있으면 staging_1의 MERGE도 함께 rollback
9

10

11
# === 분산 처리 (Spark 등) ===
12
# 두 가지 구현이 존재
13

14
# 방법 1: 태스크별 로컬 트랜잭션
15
# → 각 태스크가 독립 트랜잭션 → 잡 재시도 시 이미 커밋된 태스크 데이터 중복
16
def local_transaction_problem():
17
    # Task 1: BEGIN → INSERT → COMMIT ✓ (이미 커밋됨)
18
    # Task 2: BEGIN → INSERT → FAIL ✗
19
    # Job retry: Task 1 다시 실행 → 중복 발생!
20
    pass
21

22
# 방법 2: 잡 전체가 하나의 트랜잭션
23
# → 모든 태스크 완료 후 한 번에 커밋 → 더 강한 보장
24
def job_level_transaction():
25
    # Task 1: 파일 쓰기 (아직 커밋 안 됨)
26
    # Task 2: 파일 쓰기 (아직 커밋 안 됨)
27
    # 모든 태스크 완료 → commit log 생성 → 한 번에 공개
28
    pass

1
# Delta Lake: 커밋 로그가 생성되어야 데이터가 공개됨
2
input_data = spark.read.json("s3://input/visits/")
3

4
# 여러 태스크가 병렬로 파일을 쓰지만...
5
(input_data.write
6
    .format("delta")
7
    .mode("overwrite")
8
    .save("s3://output/visits"))
9
# → 모든 파일 쓰기 완료 후 _delta_log/에 커밋 파일 생성
10
# → 커밋 파일이 있어야 소비자가 읽을 수 있음
11
# → 커밋 전 장애 발생 시: 파일은 있지만 커밋이 없으므로 소비자에게 보이지 않음

1
# Flink에서 Kafka 트랜잭셔널 쓰기 (개념적 Python 표현)
2
kafka_sink_config = {
3
    "topic": "reduced_visits",
4
    "delivery_guarantee": "EXACTLY_ONCE",       # 트랜잭션 활성화
5
    "transaction_timeout_ms": 10 * 60 * 1000,   # 10분
6
}
7

8
# 동작 방식:
9
# 1. 프로듀서가 파티션에 트랜잭션 시작 메시지 전송
10
# 2. 데이터 레코드 전송 (소비자에게 아직 안 보임 — isolation level에 따라)
11
# 3. Flink 체크포인트 완료 시 커밋 메시지 전송
12
# 4. 이 시점부터 소비자가 데이터를 볼 수 있음
13

14
# transaction_timeout이 체크포인트보다 짧으면
15
#    트랜잭션 만료 → 커밋 불가 → 데이터 유실

1
# 두 파일의 MERGE를 하나의 트랜잭션으로 묶기
2
transaction_sql = """
3
-- 파일 1 로드 + MERGE
4
CREATE TEMPORARY TABLE changed_devices_file1 (LIKE dedp.devices);
5
COPY changed_devices_file1 FROM '/data/dataset_file1.csv' CSV DELIMITER ';' HEADER;
6
MERGE INTO dedp.devices AS d USING changed_devices_file1 AS c
7
    ON d.type = c.type AND d.version = c.version
8
    WHEN MATCHED THEN UPDATE SET full_name = c.full_name
9
    WHEN NOT MATCHED THEN INSERT (type, full_name, version) VALUES (c.type, c.full_name, c.version);
10

11
-- 파일 2 로드 + MERGE (컬럼 길이 초과 에러 발생!)
12
CREATE TEMPORARY TABLE changed_devices_file2 (LIKE dedp.devices);
13
COPY changed_devices_file2 FROM '/data/dataset_file2.csv' CSV DELIMITER ';' HEADER;
14
MERGE INTO dedp.devices AS d USING changed_devices_file2 AS c
15
    ON d.type = c.type AND d.version = c.version
16
    WHEN MATCHED THEN UPDATE SET full_name = c.full_name
17
    WHEN NOT MATCHED THEN INSERT (type, full_name, version) VALUES (c.type, c.full_name, c.version);
18

19
-- 파일 2에서 에러 → COMMIT에 도달하지 못함 → 파일 1의 MERGE도 rollback
20
COMMIT;
21
"""