1
# Kafka 토픽 생성 시 log compaction 설정
2
from confluent_kafka.admin import AdminClient, NewTopic
3

4
admin = AdminClient({"bootstrap.servers": "broker:9092"})
5

6
topic = NewTopic(
7
    "user-profiles",
8
    num_partitions=3,
9
    config={
10
        "cleanup.policy": "compact",              # compaction 활성화
11
        "min.compaction.lag.ms": "3600000",        # 최소 1시간 후 compaction
12
        "max.compaction.lag.ms": "86400000",       # 최대 24시간 내 compaction
13
    }
14
)
15
admin.create_topics([topic])

1
[Kafka Compaction 전]
2
Key=user1, Value={name: "A"}  (offset 0)
3
Key=user2, Value={name: "B"}  (offset 1)
4
Key=user1, Value={name: "C"}  (offset 2)  ← user1 업데이트
5
Key=user1, Value=null         (offset 3)  ← user1 삭제 (tombstone)
6

7
[Kafka Compaction 후]
8
Key=user2, Value={name: "B"}  (offset 1)  ← 최신 값만 유지
9
Key=user1, Value=null         (offset 3)  ← tombstone 유지 (일정 시간 후 삭제)

1
# 전략 1: 드물게 실행 (예: 하루 1회, 업무 외 시간)
2
# 낮은 컴퓨팅 비용
3
# compaction 전 데이터를 읽는 소비자는 여전히 느림
4

5
# 전략 2: 수집 파이프라인에 포함 (매번 실행)
6
# 소비자 항상 최적화된 데이터 읽기
7
# 수집 파이프라인 레이턴시 증가
8

9
# 전략 3: 적응형 — 파일 수 기반 트리거
10
def should_compact(spark: SparkSession, table_path: str, threshold: int = 500) -> bool:
11
    """파일 수가 임계값을 넘으면 compaction 트리거"""
12
    from delta.tables import DeltaTable
13

14
    dt = DeltaTable.forPath(spark, table_path)
15
    file_count = dt.detail().select("numFiles").collect()[0][0]
16
    return file_count > threshold

1
[Compaction 라이프사이클]
2
OPTIMIZE → 새 큰 파일 생성 (구 파일도 존재)
3
   ↓
4
VACUUM  → retention 지난 구 파일 삭제
5
   ↓
6
 삭제된 파일 기반 Time Travel 불가