Spark Structured Streaming

학습 목표

Structured Streaming이 unbounded 입력을 항상 켜진 job으로 처리하는 방식을 설명할 수 있다.

micro-batch 실행 모델과 trigger·checkpoint 역할을 구분할 수 있다.

event time·watermark로 늦게 도착한 이벤트를 다룰 수 있다.

window 집계와 sink output mode를 선택할 수 있다.

Kafka → Spark streaming → alert·table 파이프라인 초안을 설계할 수 있다.

문제 상황

• 5편 배치로 어제 dt 집계는 되는데 — 1분 단위 이상탐지는 cron으로 불가
• Kafka consumer lag는 쌓이는데 Spark job은 하루 한 번만 돈다
• processingTime 윈도우 썼더니 재시작 후 집계가 두 배 — checkpoint 없음
• 이벤트 timestamp와 서버 수신 시각이 다름 — 늦게 들어온 주문이 빠진 집계
• writeStream append 모드인데 집계 테이블 — sink 에러·중복 행
• 3편 micro-batch 개념만 알고 — Spark에서 어디에 설정하는지 모름

이번 편은 4편 Kafka ingest 위에서 항상 켜진 process — Structured Streaming 설계다. 배치 backfill은 5편·10편과 병행.

1. Structured Streaming이 process에서 하는 일

3편 스트림 모델 — ingest → process → sink. 5편 배치와 같은 DataFrame API로 끝나지 않는 job을 돌린다.

역할	Structured Streaming이 해주는 것
연속 읽기	Kafka·파일·socket에서 신규 데이터 증분 소비
상태 집계	window·session·join state — executor에 유지
내구성	checkpoint·WAL로 재시작·장애 복구
동일 엔진	배치와 Catalyst·shuffle — 코드 재사용

Kafka orders.events (unbounded)
        │
        ▼
Spark Structured Streaming (always-on)
        │
        ▼
alert topic · Delta table · Prometheus sink

• readStream — 소스 정의 — Kafka는 4편 consumer group·offset과 연결
• writeStream — sink + outputMode + trigger + checkpointLocation
• 배치와 분리 — T+1 정확 집계는 5편 배치, 운영 알림·근실시간 metric은 스트리밍
• Spark Streaming (DStreams) — 레거시 — 신규는 Structured Streaming만

2. Micro-batch 실행 모델

기본 모델은 micro-batch — 3편에서 말한 “스트림 API, 배치 실행”.

개념	설명
Micro-batch	trigger마다 새 데이터를 작은 배치로 읽어 DAG 실행
Trigger	얼마나 자주 micro-batch를 돌릴지 (ProcessingTime)
Continuous (옵션)	낮은 latency — 처리 모델·sink 제약 많음 — 대부분 micro-batch로 충분

events ──► buffer ──► [batch t0] [batch t1] [batch t2] ...
                         │          │          │
                      Spark job  Spark job  Spark job

• ProcessingTime trigger — trigger(processingTime="10 seconds") — 도착 기준 주기
• Once — 한 번 micro-batch 후 종료 — 테스트·bounded replay에 유용
• AvailableNow — 쌓인 backlog 한 번에 소진 후 종료 — 마이그레이션·catch-up
• Continuous processing — ms급 — Kafka 소수 sink만 — 운영 복잡 — 필요할 때만

# code/streaming_job.example.py (발췌)
query = (
    counts.writeStream
    .outputMode("update")
    .format("kafka")
    .option("checkpointLocation", "s3://ops/checkpoints/order-errors")
    .trigger(processingTime="30 seconds")
    .start()
)
query.awaitTermination()

3. Event time와 Watermark

Processing time — 워커가 읽은 시각. Event time — 이벤트 본문의 비즈니스 시각 (order_ts).

개념	설명
Event time	timestamp 컬럼 — 지연·재전송에도 의미 유지
Watermark	“이 시각 이전 event time은 더 안 온다”는 선언
Late data	watermark 넘긴 이벤트 — drop 또는 side output

event time:  ──●──●────●──────● (late)
watermark:   ─────────────|──────►
                          drop zone

• withWatermark — withWatermark("order_ts", "10 minutes") — 늦음 허용 창
• watermark 너무 짧음 — 정상 지연 데이터 유실 — 비즈니스 SLA로 정하기
• watermark 너무 김 — state·checkpoint 크기 증가 — 메모리·복구 시간
• Kafka — producer clock skew — event time은 소스 신뢰도 검토

# event time 기반 윈도우
events = (
    kafka_df
    .withWatermark("order_ts", "15 minutes")
    .groupBy(window("order_ts", "5 minutes"), col("status"))
    .count()
)

4. Window와 Trigger

Window — event time 구간별 집계. Trigger는 언제 결과를 sink에 밀어 넣을지.

윈도우	설명	예
Tumbling	겹치지 않는 고정 구간	5분 에러 건수
Sliding	겹치는 구간 + slide	1시간 평균, 10분마다 갱신
Session	갭 기준 세션	사용자 활동 구간

Trigger	동작
Default (micro-batch)	가능한 빨리 — 보통 ~100ms 간격
ProcessingTime	고정 주기 — 부하·비용 예측 용이
Continuous	지연 최소화 — sink·연산 제한

• window + watermark — 완료된 tumbling window만 finalize — late drop과 연동
• Trigger ≠ window — 5분 window + 30초 trigger → 같은 window 결과를 여러 번 update sink
• 5편 shuffle — window 집계도 groupBy — shuffle.partitions 튜닝 동일

5. Sink와 Checkpoint

Checkpoint — offset·state·WAL — exactly-once에 가깝게 재시작. Sink mode는 집계 유형과 맞춘다.

outputMode	의미	적합한 쿼리
append	새 행만 추가	비집계 또는 watermark 있는 event-time window
update	변경된 키만 갱신	집계·mapGroupsWithState
complete	전체 결과 테이블 매번	작은 집계 (top-N)

checkpoint/
  offsets/     ← Kafka 등 소스 위치
  state/       ← window·join state
  commits/     ← 완료된 batch 메타

• checkpointLocation — 고유·내구 스토리지 (S3·HDFS) — 삭제 시 처음부터 재처리
• Kafka sink — topic·kafka format — at-least-once — downstream idempotent 설계
• Delta·Iceberg — merge sink — 8편 lakehouse와 연결
• foreachBatch — micro-batch마다 임의 로직 (JDBC upsert 등) — 유연·운영 책임 증가

# code/streaming-job.conf.yaml (발췌)
streaming:
  checkpointLocation: s3://ops/checkpoints/payment-errors
  trigger: processingTime
  triggerInterval: 30s
  outputMode: update
  kafka:
    subscribe: orders.events
    startingOffsets: latest
  watermark: 15m

6. 스트리밍 job 설계 체크리스트

#	질문
1	latency SLA가 배치로 안 되는가 — micro-batch 주기로 맞는가
2	event time 컬럼·watermark 지연이 비즈니스와 맞는가
3	checkpoint 경로가 영구·job별 고유인가
4	outputMode가 집계 유형과 맞는가
5	consumer group·offset 정책 (latest vs earliest)이 재처리 요구와 맞는가
6	배치 gold와 스트림 metric 이중 집계 시 정의가 충돌하지 않는가

# Kafka ingest → 1분 에러율 alert (개요)
stream = (
    spark.readStream.format("kafka")
    .option("subscribe", "orders.events")
    .option("startingOffsets", "latest")
    .load()
)
parsed = stream.select(from_json(col("value").cast("string"), schema).alias("data")).select("data.*")
errors = (
    parsed.withWatermark("order_ts", "10 minutes")
    .groupBy(window("order_ts", "1 minute"))
    .agg(
        spark_sum(when(col("status") == "ERROR", 1).otherwise(0)).alias("err"),
        count("*").alias("total"),
    )
)

• Lag 모니터링 — Kafka consumer lag + Spark batch duration — trigger보다 batch가 길면 쌓임
• Replay — checkpoint 삭제 + earliest — 4편 retention 창 안에서만
• 정확한 일일 매출 — 5편 배치 유지, 스트림은 운영·근실시간만
• 7편 — warehouse 적재는 ETL vs ELT와 역할 분담

정리

• Structured Streaming — Kafka 등 unbounded ingest 위 항상 켜진 process
• Micro-batch — trigger마다 작은 배치 — 대부분의 실무 선택
• Event time + watermark — 늦은 이벤트와 state 크기 균형
• Window + trigger — 무엇을 집계 vs 얼마나 자주 sink
• Checkpoint + outputMode — 재시작·집계 sink 호환 — 배치 gold와 역할 분리

다음에 다룰 것

• ETL vs ELT
• warehouse loading, dbt, orchestration

---

해당 내용은 Spark Documentation (Structured Streaming Guide), Kleppmann — Designing Data-Intensive Applications (Ch 11) 를 기반으로 합니다.