PySpark Kafka 스트리밍 심화 — 오프셋, 백프레셔, Exactly-Once

컨베이어 벨트를 상상해 보세요. 수백만 개의 택배 상자가 쉬지 않고 흘러오고, 여러분은 그것을 하나씩 꺼내 창고에 쌓아야 합니다. 벨트가 멈춰도, 다시 시작했을 때 "어디까지 집었는지"를 정확히 기억해야 상자를 빠뜨리거나 두 번 집지 않겠죠. Kafka 스트리밍이 딱 그 구조입니다 — 벨트는 Kafka, 창고는 Lakehouse, "어디까지 집었는지"는 오프셋과 체크포인트가 책임집니다.

Kafka 는 실시간 데이터의 사실상 표준 버스이고, Spark Structured Streaming 은 이를 소비해 Lakehouse 에 적재하는 가장 흔한 도구입니다. 그런데 "Kafka 에서 읽어서 쓴다"는 단순해 보이는 작업에 — 오프셋 관리, 백프레셔, 스키마 파싱, exactly-once, 컨슈머 랙 — 운영의 디테일이 잔뜩 숨어 있습니다.

이 글은 PySpark 로 Kafka 를 안정적으로 소비·적재하는 실전 패턴을 정리합니다. (스트리밍 상태 관리·워터마크 일반론은 별도 글 "PySpark Structured Streaming 상태 관리와 Exactly-Once"를 참고하세요.)

이 글에서 배우는 것

• Kafka 소스를 읽는 기본 옵션과 메시지 구조
• 체크포인트로 오프셋을 관리하는 원리
• maxOffsetsPerTrigger 로 OOM 을 막는 백프레셔 설정
• from_json/from_avro 로 바이너리 메시지를 구조화하는 방법
• foreachBatch + 멱등 MERGE 로 exactly-once 를 달성하는 패턴
• 컨슈머 랙 모니터링과 안전한 재처리 방법

1. Kafka 소스 읽기 기본

Kafka 소스를 읽을 때 설정해야 할 옵션은 많지 않지만, 각각이 운영에 직결됩니다. 어떤 브로커에 연결할지, 어떤 토픽을 구독할지, 그리고 처음 실행 시 어디서부터 읽을지를 지정해 주면 됩니다.

from pyspark.sql import functions as F
 
raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
    .option("subscribe", "events")                  # 토픽
    .option("startingOffsets", "latest")            # 또는 earliest
    .option("maxOffsetsPerTrigger", "500000")       # 백프레셔(아래)
    .load())
 
# Kafka 메시지는 바이너리 — key/value 는 binary
# raw 의 컬럼: key, value, topic, partition, offset, timestamp

2. 오프셋 관리 — 체크포인트가 전부

많은 분들이 "Kafka 컨슈머 그룹이 오프셋을 관리하지 않나요?"라고 묻습니다. Structured Streaming 에서는 그렇지 않습니다. "어디까지 읽었나"(오프셋)를 체크포인트가 관리합니다. Kafka 컨슈머 그룹의 __consumer_offsets 가 아니라, Spark 체크포인트가 진실의 원천입니다.

query = (parsed.writeStream
    .option("checkpointLocation", "s3://bucket/ckpt/events")  # 필수
    .start())

핵심: startingOffsets 는 체크포인트가 없는 첫 실행에만 적용됩니다. 재시작 시에는 체크포인트의 오프셋에서 이어갑니다. 그래서 체크포인트를 지우면 처음부터(또는 latest 부터) 다시 읽게 되니 주의하세요. 체크포인트는 오프셋·상태·진행상황을 모두 담은 핵심 자산입니다.

3. 백프레셔 — maxOffsetsPerTrigger

컨베이어 벨트가 몇 시간 멈춰 있다가 갑자기 돌아오면 어떻게 될까요? 밀린 상자가 한꺼번에 쏟아집니다. Spark 스트리밍도 마찬가지입니다 — 스트리밍이 한참 멈췄다 재개되거나 트래픽이 폭증하면, 한 마이크로배치가 수천만 건을 한 번에 처리하려다 OOM 이 납니다. maxOffsetsPerTrigger 로 배치당 최대 메시지 수를 제한합니다.

.option("maxOffsetsPerTrigger", "500000")   # 트리거당 최대 50만 건

백프레셔 없음: 밀린 1억 건을 한 배치에 → OOM 💥
백프레셔 있음: 배치당 50만 건씩 나눠 처리 → 안정적으로 따라잡음

이 값은 배치 처리 능력에 맞춰 정하면 됩니다. 너무 작으면 지연(랙)이 안 줄고, 너무 크면 배치가 무거워지니 — 처리 시간을 모니터링하며 조금씩 올리는 식으로 튜닝하세요.

팁: 적재 초기에는 보수적으로 시작(예: 10만 건)하고, StreamingQueryProgress 의 처리 시간이 트리거 간격의 절반 이하로 유지되면 값을 올려 보세요.

4. 스키마 파싱 — binary value → 구조화

Kafka 에서 꺼낸 메시지의 value 는 날 것의 바이너리입니다. 이걸 그대로 쓸 수는 없으니, 스키마를 선언하고 from_json 또는 from_avro 로 구조화된 컬럼으로 변환해야 합니다. Kafka value 는 보통 JSON 또는 Avro 입니다.

from pyspark.sql import types as T
 
schema = T.StructType([
    T.StructField("user_id", T.LongType()),
    T.StructField("event_type", T.StringType()),
    T.StructField("event_time", T.TimestampType()),
])
 
parsed = (raw
    .select(
        F.col("key").cast("string").alias("key"),
        F.from_json(F.col("value").cast("string"), schema).alias("data"),
        F.col("timestamp").alias("kafka_ts"),
        F.col("offset"))
    .select("key", "data.*", "kafka_ts", "offset"))

Avro + Schema Registry 를 쓰면 from_avro 와 레지스트리 연동을 사용합니다. (Avro 스키마 일반론은 별도 글 "Apache Avro Schema 완벽 가이드" 참고.) 운영 환경에서는 언제나 파싱에 실패하는 "깨진 메시지"가 들어오므로, 격리 패턴으로 따로 보관해 두세요(별도 글 "중첩 반정형 데이터"의 quarantine 패턴).

한 문장으로: from_json 으로 바이너리를 구조화하고, 파싱 실패 행은 격리해 파이프라인이 멈추지 않게 합니다.

5. Exactly-Once 적재 — foreachBatch + MERGE

"한 번만 처리"는 스트리밍에서 가장 어려운 보장입니다. 재시작하면 같은 메시지를 다시 읽을 수 있고, 그 경우 중복 적재가 생기면 안 됩니다. Kafka→Lakehouse 적재에서 exactly-once 는 체크포인트(재생 가능 소스) + 멱등 싱크의 조합으로 성립합니다. 파일 append 대신 foreachBatch 로 Iceberg/Delta MERGE 를 씁니다.

def upsert(batch_df, batch_id):
    # 배치 내 키 중복 제거 (MERGE source 는 키당 1행)
    from pyspark.sql.window import Window
    w = Window.partitionBy("user_id", "event_time").orderBy(F.col("offset").desc())
    deduped = batch_df.withColumn("rn", F.row_number().over(w)).where("rn=1").drop("rn")
    deduped.createOrReplaceTempView("u")
 
    batch_df.sparkSession.sql("""
        MERGE INTO analytics.events t USING u s
        ON t.user_id = s.user_id AND t.event_time = s.event_time
        WHEN NOT MATCHED THEN INSERT *
    """)
 
query = (parsed.writeStream
    .option("checkpointLocation", "s3://bucket/ckpt/events")
    .foreachBatch(upsert)
    .trigger(processingTime="1 minute")
    .start())

재시작으로 같은 배치가 재처리돼도, MERGE 의 키 매칭으로 중복이 안 생깁니다(멱등). Kafka offset 을 dedup 키나 타이브레이커로 활용하면 정합성이 더 견고해집니다.

한 문장으로: 체크포인트가 "어디까지 읽었나"를 보장하고, 멱등 MERGE 가 "중복 없이 썼나"를 보장해 — 둘이 합쳐져 exactly-once 가 됩니다.

6. 트리거 모드와 작은 파일

트리거 간격이 짧으면(5초) 작은 파일이 폭증합니다. 적재 스트리밍은 보통 1~5분 트리거로 배치를 키우고, 그래도 쌓이는 작은 파일은 정기 컴팩션으로 정리합니다(별도 글 "PySpark Small Files Problem"). availableNow 는 "스트리밍 코드로 주기 배치"를 돌리는 실용적 모드입니다.

7. 멀티 토픽·파티션과 병렬성

여러 토픽을 동시에 구독하거나, 패턴으로 토픽을 선택하는 것도 간단하게 설정할 수 있습니다. 다만 처리량을 높이려면 Spark 설정만으로는 부족하고, Kafka 파티션 수가 병렬성의 상한을 결정한다는 점을 기억하세요.

.option("subscribe", "events,clicks,orders")      # 여러 토픽
.option("subscribePattern", "logs-.*")            # 패턴 구독

• 병렬성은 Kafka 파티션 수에 묶입니다. 토픽 파티션이 적으면 Spark task 도 그만큼만 — 처리량이 안 늘면 토픽 파티션을 늘려야 할 수 있습니다.
• minPartitions 로 Spark 쪽 파티션을 더 쪼개 병렬도를 높일 수 있습니다(파티션당 데이터가 클 때).

8. 운영 — 컨슈머 랙과 재처리

파이프라인을 배포한 뒤의 실제 운영 과제는 크게 두 가지입니다 — "지금 잘 따라가고 있나"(랙 모니터링)와 "과거 데이터를 다시 처리해야 할 때"(재처리)입니다.

# 진행 상황 모니터링
for q in spark.streams.active:
    print(q.lastProgress)   # 배치 시간, 입력 속도, 처리 속도, 랙

재처리(특정 시점부터 다시): 기존 체크포인트를 건드리지 말고 새 체크포인트 경로 + startingOffsets 로 별도 실행하세요. 기존 체크포인트를 지우면 진행상황·상태가 사라집니다.

9. 정리

Kafka 스트리밍의 핵심은 "체크포인트가 오프셋·상태·진행상황을 모두 관리한다"는 사실을 중심에 두는 것입니다. 백프레셔로 폭주를 막고, from_json 으로 안전하게 파싱하며, foreachBatch + 멱등 MERGE 로 exactly-once 를 달성하면 — 24시간 흐르는 실시간 파이프라인을 중복·유실 없이 운영할 수 있습니다. 트리거 간격과 컴팩션으로 작은 파일까지 다스리면 Lakehouse 실시간 적재가 완성됩니다.

마치며 — 핵심 요약

• 체크포인트가 진실의 원천입니다. Kafka 컨슈머 그룹 오프셋이 아니라 Spark 체크포인트가 오프셋·상태·진행상황을 모두 담습니다. 체크포인트 경로를 잃으면 처음부터 다시 읽게 됩니다.
• maxOffsetsPerTrigger 로 OOM 을 막으세요. 밀린 데이터가 한 배치에 몰리지 않도록 상한을 반드시 설정하고, 처리 시간을 보며 점진적으로 조정하면 됩니다.
• 바이너리 value 는 from_json/from_avro 로 구조화하고, 파싱 실패는 격리하세요. 깨진 메시지 하나가 전체 스트림을 멈추게 두면 안 됩니다.
• Exactly-once 는 멱등 MERGE 로 완성됩니다. 체크포인트가 소스 측을 보장하고, foreachBatch + MERGE 가 싱크 측 중복을 막아야 진정한 exactly-once 입니다.
• 재처리는 새 체크포인트 경로로. 기존 체크포인트를 지우거나 덮어쓰면 상태와 진행상황을 잃습니다. 재처리 시에는 항상 별도 경로와 startingOffsets 를 지정하세요.
• 병렬성의 상한은 Kafka 파티션 수입니다. 처리량이 안 늘어난다면 Spark 설정보다 토픽 파티션을 늘리는 것을 먼저 검토하세요.

컨베이어 벨트가 하루 24시간 멈추지 않고 돌아가듯, 여러분의 Kafka-Lakehouse 파이프라인도 이 패턴들을 갖추면 중단·중복·유실 없이 안정적으로 흘러갈 수 있습니다.

이 글은 Spark 3.5 + Kafka 기준으로 작성되었습니다. 실시간 Kafka-Lakehouse 적재 파이프라인 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀