5편 · 약 17분

구조적 스트리밍(Structured Streaming)

배치와 스트리밍의 경계가 사라지다

Spark가 처음 등장했을 때 실시간 데이터를 처리하는 방법은 DStream(Discretized Stream)이었다. RDD를 짧은 시간 단위로 묶어 배치처럼 처리하는 방식이었는데, 개발자가 배치 코드와 스트리밍 코드를 따로 작성해야 했고, 이벤트 시간 기반 처리나 지연 데이터 핸들링은 직접 구현해야 하는 영역이었다.

Spark 2.0부터 등장한 Structured Streaming은 다른 방향을 택했다. "스트리밍을 배치 쿼리가 실행되는 무한히 커지는 테이블"로 추상화한 것이다. 개발자는 DataFrame/Dataset API 또는 Spark SQL로 쿼리를 작성하고, 그 쿼리가 새 데이터가 도착할 때마다 점진적으로 실행된다. 코드 관점에서는 배치와 스트리밍이 같은 API를 쓴다.

이 추상화가 실용적인 이유는 단순히 API 편의성 때문만이 아니다. Catalyst 옵티마이저, AQE, 풍부한 내장 함수가 스트리밍 쿼리에도 그대로 적용된다. 그리고 엔진이 이벤트 시간, 워터마크, 상태 관리를 책임지기 때문에 개발자는 "어떤 데이터를 어떻게 집계할지"에 집중할 수 있다.

Structured Streaming 실행 모델

무한 테이블 모델: 스트리밍을 배치처럼 생각하기

Structured Streaming의 핵심 아이디어는 입력 데이터를 계속 행이 추가되는 테이블로 보는 것이다. 이 테이블에 Spark SQL 쿼리를 실행하면, 쿼리 결과는 Result Table이 되고, 그 Result Table을 외부 시스템에 쓰는 것이 스트리밍 출력이다.

실제 실행 방식은 기본적으로 마이크로배치다. 엔진은 Trigger가 발생할 때마다 새로 도착한 데이터를 읽어 이전 상태와 병합하고, Result Table의 변경 부분을 싱크에 쓴다. 각 트리거 구간은 일반 Spark 배치 작업처럼 실행된다. 따라서 Catalyst 최적화, Spark UI 통한 모니터링, checkpointing 기반 재실행이 모두 자연스럽게 적용된다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, count

spark = SparkSession.builder.appName("streaming-demo").getOrCreate()

# 입력 소스: Kafka
events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "page-events")
    .load()
    .selectExpr("CAST(value AS STRING) AS json",
                "timestamp AS event_time"))

# 쿼리: 1분 tumbling window로 이벤트 수 집계
agg = (events
    .groupBy(window("event_time", "1 minute"), col("json"))
    .count())

# 출력 싱크: 콘솔 (개발용)
query = (agg.writeStream
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "/tmp/checkpoints/demo")
    .trigger(processingTime="10 seconds")
    .start())

query.awaitTermination()

출력 모드: 어떤 행을 싱크로 보낼 것인가

Result Table 전체를 매 trigger마다 쓰면 비효율적이다. 그래서 Structured Streaming은 세 가지 출력 모드를 제공한다.

출력 모드	무엇을 씀	주로 쓰는 상황	제한
append	이번 트리거에서 확정된 새 행만	집계 없이 개별 이벤트를 쓸 때 / 워터마크로 확정된 윈도우 결과	집계가 있으면 워터마크 필수
update	변경되거나 새로 생긴 행만	집계 결과를 지속적으로 갱신하는 HBase/DB 싱크	Delta를 제외하면 지원 싱크 제한적
complete	Result Table 전체	전체 집계 결과를 매번 덮어쓸 때 (작은 테이블)	데이터가 커질수록 비용 증가

실무에서 가장 자주 쓰는 조합은 Kafka나 파일 싱크에 append 모드다. 집계가 있으면 워터마크로 윈도우를 먼저 확정해야 append 모드를 쓸 수 있다.

트리거: 처리 간격을 결정하는 방법

트리거는 엔진이 새 마이크로배치를 언제 시작할지 결정한다.

트리거 종류	설정 예시	동작	언제 쓰나
`processingTime`	`"10 seconds"`	고정 주기로 마이크로배치 실행	일반 실시간 스트리밍
`once`	`Trigger.Once()`	현재 사용 가능한 데이터 전부 처리 후 종료	배치-스트리밍 하이브리드
`availableNow`	`Trigger.AvailableNow()`	`once`와 유사하지만 여러 마이크로배치로 나눠 처리	대량 미처리 데이터 효율적 처리
`continuous`	`Trigger.Continuous("1 second")`	레코드 단위 준-실시간 처리 (실험적)	저지연 요구, Kafka 소스 한정

availableNow는 Spark 3.3+에서 추가됐다. once가 모든 데이터를 하나의 거대한 배치로 처리하다 OOM이 날 수 있는 문제를 해결한다. 주기적으로 실행되는 스케줄 작업에서 스트리밍 체크포인트의 증분 이점을 누리고 싶을 때 실용적이다.

상태 기반 연산: 윈도우 집계와 워터마크

스트리밍의 진짜 어려움은 이벤트 시간(event time)과 처리 시간(processing time)이 다르다는 점이다. 모바일 앱이 오프라인 상태였다가 네트워크가 연결되면, 몇 분 전 이벤트가 지금 도착할 수 있다. 이때 "1분 윈도우"를 이벤트가 도착한 시각 기준으로 집계하면 올바른 결과를 얻을 수 없다.

윈도우 집계

Structured Streaming은 이벤트 타임스탬프 컬럼을 기준으로 윈도우를 정의한다.

Tumbling window: 겹치지 않는 고정 크기 구간. window("event_time", "5 minutes")
Sliding window: 겹치는 이동 구간. window("event_time", "10 minutes", "5 minutes") → 5분마다 슬라이딩하는 10분 윈도우
Session window: 이벤트 간 비활동 간격으로 세션을 정의. Spark 3.2+에서 지원

워터마크: 지연 데이터를 얼마나 기다릴 것인가

윈도우마다 상태를 무한히 유지하면 메모리가 넘친다. 워터마크는 "이 시점보다 오래된 이벤트는 버려도 된다"는 기준을 이벤트 시간 기준으로 정의한다.

from pyspark.sql.functions import window, col

windowed = (events
    .withWatermark("event_time", "2 minutes")   # 최대 2분 지연 허용
    .groupBy(
        window("event_time", "1 minute"),
        col("user_id")
    )
    .count())

withWatermark("event_time", "2 minutes")의 의미:

엔진이 지금까지 본 최대 이벤트 시간 = max_seen_event_time
현재 워터마크 = max_seen_event_time − 2분
워터마크보다 오래된 윈도우는 확정되고 상태에서 제거됨
워터마크보다 이전 시간의 지연 이벤트는 해당 윈도우가 이미 닫혔으므로 무시됨

워터마크와 윈도우 확정 타이밍

워터마크의 실무 함정은 소스가 일시적으로 멈출 때다. 카프카 파티션이나 파일 소스에서 이벤트가 들어오지 않으면 max_seen_event_time이 갱신되지 않고, 워터마크도 전진하지 않는다. 그 결과 오래된 윈도우가 상태에 계속 남아 State Store가 팽창한다. 운영 환경에서는 소스별 이벤트 흐름을 모니터링하고, 너무 긴 워터마크 지연은 메모리 압박의 신호로 봐야 한다.

스트림-스트림 조인과 임의 상태 관리

Structured Streaming은 두 스트림을 이벤트 시간 기준으로 조인할 수 있다. 예를 들어 ad-impressions 스트림과 ad-clicks 스트림을 ad_id와 이벤트 시간 범위로 조인해 광고 클릭률을 실시간으로 계산할 수 있다.

# 두 스트림에 워터마크 설정 (stream-stream join 필수)
impressions = (spark.readStream.format("kafka")
    .option("subscribe", "ad-impressions").load()
    .withWatermark("imp_time", "2 hours"))

clicks = (spark.readStream.format("kafka")
    .option("subscribe", "ad-clicks").load()
    .withWatermark("click_time", "3 hours"))

# 이벤트 시간 범위 조인: 클릭은 노출 후 1시간 이내
from pyspark.sql.functions import expr
joined = impressions.join(
    clicks,
    expr("""
        imp_id = click_id AND
        click_time >= imp_time AND
        click_time <= imp_time + interval 1 hour
    """),
    "leftOuter"
)

더 복잡한 커스텀 상태 처리가 필요할 때는 flatMapGroupsWithState나 mapGroupsWithState를 쓴다. 사용자 세션 트래킹, 머신의 상태 전이 추적 같은 시나리오에서 유용하다. 단, 상태 관리 로직을 직접 작성해야 하고, 상태 타입의 Encoder도 명시해야 한다는 점이 복잡하다.

체크포인팅과 정확히 한 번 처리

Structured Streaming이 장애 후 재시작해도 데이터를 잃거나 중복 처리하지 않으려면 두 조건이 충족되어야 한다.

재실행 가능한(replayable) 소스: Kafka, 파일처럼 오프셋이나 경로로 재읽기 가능한 소스
멱등적(idempotent) 싱크 또는 트랜잭션 지원 싱크: Delta Lake, 고유키 기반 DB 업서트, Kafka의 exactly once 설정

체크포인트 디렉터리에는 두 종류의 정보가 저장된다.

오프셋 로그: 마지막으로 처리한 소스 오프셋 (다음 배치 시작점)
커밋 로그: 싱크에 쓰기까지 완료됐다는 확인 (재실행 시 중복 쓰기 방지)

/checkpoint/
  offsets/     ← 각 트리거의 소스 오프셋
  commits/     ← 완료된 트리거 번호
  state/       ← State Store 스냅샷 (집계·조인 상태)
  metadata     ← 쿼리 메타데이터

체크포인트 디렉터리를 변경하거나 삭제하면 상태가 리셋된다. 운영 중인 스트리밍 쿼리의 소스·싱크·스키마를 크게 바꾸면 체크포인트 비호환이 발생할 수 있다. 그런 변경이 필요할 때는 체크포인트를 새로 만들거나, Spark 문서에서 해당 변경이 허용 가능한지 확인해야 한다.

Structured Streaming과 DStream 비교

DStream을 아직 쓰는 코드를 마이그레이션하거나, 두 API의 차이를 설명해야 할 때를 위해 차이를 정리해 두자.

항목	DStream (레거시)	Structured Streaming
API 수준	RDD 기반, 저수준	DataFrame/Dataset, SQL
최적화	없음 (사용자가 수동 최적화)	Catalyst + AQE
이벤트 시간 처리	직접 구현	내장 (window + watermark)
정확히 한 번 처리	소스/싱크 의존, 복잡	체크포인팅 + 재실행 가능 소스로 내장
상태 관리	직접 관리	State Store 내장
Spark 3.x 지원	유지보수 모드	주력 API

DStream은 Spark 3.x에서 기능 추가 없이 유지보수만 되고 있다. 신규 프로젝트는 Structured Streaming으로 시작하고, 기존 DStream 코드는 마이그레이션 기회가 생길 때 전환하는 것이 바람직하다.

운영 모니터링: StreamingQueryListener

Structured Streaming 쿼리의 상태를 프로그래밍 방식으로 수집하려면 StreamingQueryListener를 등록한다.

from pyspark.sql.streaming import StreamingQueryListener

class MyListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        print(f"Query started: {event.id}")

    def onQueryProgress(self, event):
        p = event.progress
        print(f"Batch {p.batchId}: "
              f"inputRowsPerSecond={p.inputRowsPerSecond:.1f}, "
              f"processedRowsPerSecond={p.processedRowsPerSecond:.1f}, "
              f"numInputRows={p.numInputRows}, "
              f"statefulOperator stateRows={[s.numRowsTotal for s in p.stateOperators]}")

    def onQueryTerminated(self, event):
        print(f"Query terminated: {event.id}")

spark.streams.addListener(MyListener())

onQueryProgress의 StreamingQueryProgress 객체에는 배치당 입력 행 수, 처리 행 수, 처리 지연 시간(trigger duration), 워터마크 값, 상태 행 수, 소스별 오프셋 등이 담긴다. 이 값을 Prometheus Pushgateway나 내부 메트릭 시스템으로 보내면 스트리밍 지연과 처리량을 대시보드에서 실시간으로 볼 수 있다.

Spark UI의 Structured Streaming 탭에서는 실행 중인 쿼리의 inputRowsPerSecond, processedRowsPerSecond, batchDuration 그래프를 볼 수 있다. 처리 속도가 입력 속도를 따라가지 못하면 지연이 누적되고 있다는 신호다.

마무리

Structured Streaming은 "스트리밍을 무한히 커지는 배치 테이블"로 추상화함으로써, 배치와 스트리밍을 같은 API와 최적화 엔진으로 처리하게 만들었다. 출력 모드와 트리거로 처리 방식을 조절하고, 워터마크로 이벤트 시간 지연 데이터를 다루며, 체크포인팅으로 장애 내구성을 확보한다.

Structured Streaming을 쓸 때 중요한 습관은 세 가지다. 첫째, 워터마크를 명시하지 않으면 상태가 무한히 쌓인다. 둘째, 체크포인트 디렉터리는 영구 스토리지에 두고 함부로 바꾸지 않는다. 셋째, 처리 속도와 상태 크기를 지속적으로 모니터링한다. 이 세 가지만 지켜도 Structured Streaming 운영의 흔한 장애 상당수를 예방할 수 있다.