3. Checkpoints

3.1 The Problem: Snapshotting a Moving Stream

스트림 job에서는 여러 머신에 흩어진 operator들이 각자 상태(집계 값, window 내용)를 가지고 끊임없이 레코드를 주고받는다. 장애가 나면 모든 operator의 상태와 source가 읽은 위치가 서로 맞는 시점으로 되돌아가야 정확히 한 번 처리한 결과가 나온다.

가장 단순한 방법은 전체를 멈추고 스냅샷을 찍는 것이지만, 스트림 처리에서는 받아들일 수 없다. 고전적인 분산 스냅샷 알고리즘인 Chandy-Lamport는 멈추지 않고 스냅샷을 찍지만, 채널 위를 지나가는 중인 메시지까지 저장해야 한다. Flink의 ABS(Asynchronous Barrier Snapshotting)는 dataflow가 방향이 있는 그래프라는 점을 이용해 이 저장을 피한다.

3.2 Barriers

barrier는 스트림 속에 끼워 넣는 구분선이다.

sequenceDiagram
  participant JM as JobManager
  participant SRC as source
  participant OP as operator
  participant SNK as sink
  participant ST as checkpoint 저장소
  JM->>SRC: checkpoint n 시작
  SRC->>ST: 읽은 위치 저장
  SRC->>OP: 레코드 뒤에 barrier n
  OP->>ST: 상태 스냅샷
  OP->>SNK: barrier n 전달
  SNK->>ST: 상태 스냅샷
  ST-->>JM: 모든 operator 완료
  JM->>JM: checkpoint n 확정

1. JobManager의 checkpoint coordinator가 source에 checkpoint n을 시작하라고 알린다.
2. source는 지금 읽은 위치(Kafka offset 등)를 저장하고, 스트림 안에 barrier n이라는 특별한 레코드를 끼워 보낸다.
3. barrier는 레코드를 앞지르지 않고 데이터와 같은 순서로 흐른다. 그래서 barrier 앞의 레코드는 checkpoint n에 속하고, 뒤의 레코드는 n+1에 속한다.
4. operator가 barrier n을 받으면 자기 상태의 스냅샷을 찍고, barrier를 다음 operator로 내보낸다.
5. sink까지 모든 operator가 스냅샷을 마치고 보고하면 JobManager가 checkpoint n을 완료로 표시한다.

스냅샷 시점에는 barrier 이전 레코드가 모두 상태에 반영됐고, 이후 레코드는 하나도 반영되지 않았다. 그래서 채널 위의 메시지를 따로 저장할 필요가 없다. 복구할 때 source를 저장된 위치부터 다시 읽으면 그 메시지들이 다시 흐른다.

이 방식은 source가 다시 읽을 수 있어야(replayable) 성립한다. Kafka 같은 내구 로그가 Flink source로 잘 맞는 것도 이 때문이다.

3.3 Barrier Alignment

operator에 입력 채널이 여러 개 있으면(join이나 keyBy 뒤의 operator) barrier n이 채널마다 다른 시각에 도착한다. barrier가 먼저 온 채널의 레코드를 계속 처리하면, checkpoint n 이후의 레코드가 상태에 섞여 들어간다.

그래서 operator는 barrier가 먼저 온 채널을 막고, 다른 모든 채널에서도 barrier n이 올 때까지 기다린다. 이것이 barrier alignment다. 모든 채널의 barrier가 모이면 스냅샷을 찍고 막았던 채널을 다시 연다.

input A:  a1 a2 [n] a3 a4
input B:  b1 b2 b3 b4 ........ [n]   ← 늦게 옴

A의 barrier n이 도착하면 a3, a4는 잠시 대기
B의 barrier n까지 오면 snapshot n을 찍고 a3, a4 처리 재개
→ snapshot n에는 barrier 앞의 레코드만 반영

alignment는 정확성을 보장하지만 대가가 따른다. 한 채널이 느리면(backpressure, 데이터 쏠림) 나머지 채널이 그만큼 막혀 지연이 늘고, checkpoint가 오래 걸린다. 이 대가를 피하는 두 가지 선택지가 있다. 하나는 alignment를 하지 않는 at-least-once mode이고, 다른 하나는 3.9에서 볼 unaligned checkpoint다. at-least-once mode가 정확히 무엇을 포기하는지는 4장에서 Kafka와 함께 본다.

3.4 Asynchronous Snapshots

ABS의 A는 스냅샷을 쓰는 동안 처리를 멈추지 않는다는 뜻이다. operator는 상태의 복사본이나 copy-on-write 참조만 빠르게 만들어 두고 처리를 계속한다. S3나 HDFS 같은 내구 저장소에 실제로 쓰는 일은 background에서 한다. 상태가 클 때 이 부분이 얼마나 가벼운지는 state backend에 달려 있다(2장).

3.5 What a Checkpoint Contains

checkpoint는 operator state와 input 위치를 같은 시점 기준으로 함께 저장한다. Kafka를 읽는 job이라면 input 위치는 partition별 offset이다. checkpoint 하나를 열어 보면 대략 이런 모양이다.

checkpoint 42
├─ KafkaSource     partition 0 → offset 1000, partition 1 → offset 980
├─ window operator key별 집계 중간값
└─ KafkaSink       아직 commit하지 않은 transaction 정보

이 세 줄이 모두 같은 순간을 가리킨다는 점이 핵심이다. 그래서 장애가 나도 42번 시점으로 한 번에 깔끔하게 되돌아갈 수 있다. 4장과 5장도 모두 이 그림에서 출발한다.

3.6 Checkpoint Is Not a Micro-batch, Nor a Window

checkpoint 주기와 window 크기는 서로 묶여 있지 않다. 10초마다 checkpoint를 찍는다고 window가 10초가 되는 것이 아니다. checkpoint는 처리를 나누는 단위도 아니다.

Spark Streaming
  epoch = 실행 단위 (micro-batch 하나를 모아서 처리)

Flink
  record = 실행 단위
  epoch  = 스냅샷 일관성 경계

Flink에서 레코드는 checkpoint와 상관없이 계속 흐르고, barrier는 그 흐름 위에 “이 지점까지를 하나의 스냅샷으로 본다"는 경계만 긋는다. 그래서 exactly-once는 물리적으로 한 번만 실행된다는 뜻이 아니다. 장애가 나면 상태를 스냅샷으로 되돌리고 source offset을 되감아 다시 처리한다. 레코드는 두 번 실행될 수 있지만, 결과와 상태에는 한 번만 반영된다.

3.7 Interval, Pause, Timeout

checkpoint는 기본적으로 꺼져 있다. execution.checkpointing.interval을 정해야 비로소 켜진다. 주기와 관련된 key 네 개는 서로 맞물려 움직인다. interval이 60초, min-pause가 30초일 때를 그려 보면 이렇다.

정상일 때 (checkpoint가 5초 걸림)
0s        60s       120s
|ckpt|....|ckpt|....|ckpt|....

느릴 때 (checkpoint가 50초 걸림)
0s                  80s               ...
|ckpt..........50s|쉼 30s|ckpt|....
                   ↑ min-pause 때문에 60초가 아니라 80초에 시작

• interval은 checkpoint를 시작하는 간격이다. 짧게 잡을수록 장애 때 다시 처리할 양이 줄지만 저장 부하가 늘어난다.
• min-pause는 checkpoint 하나가 끝나고 다음 checkpoint를 시작하기 전에 최소한 쉬어야 하는 시간이다. checkpoint가 오래 걸릴 때 job이 쉬지 않고 checkpoint만 하는 상황을 막는다.
• timeout을 넘긴 checkpoint는 실패로 보고 버린다.
• tolerable-failed-checkpoints는 checkpoint가 연속으로 몇 번까지 실패해도 괜찮은지 정한다. 기본값 0은 한 번만 실패해도 job이 재시작된다는 뜻이라 생각보다 엄격하다.

3.8 Storage and Retention

3.9 Large State and Backpressure

checkpoint가 느려지는 원인은 크게 두 가지다. 저장할 상태가 크거나, barrier가 늦게 도착하는 경우다.

상태가 크다면 2장의 증분 checkpoint로 변경분만 올린다.

barrier가 늦게 오는 것은 대개 backpressure 때문이다. 다음 operator가 느려서 buffer에 데이터가 쌓이면, barrier도 그 데이터 뒤에서 차례를 기다려야 한다. 이때 unaligned checkpoint를 켜면 barrier가 buffer에 쌓인 데이터를 앞질러 간다. 대신 앞질러 간 buffer의 데이터(in-flight 데이터)를 checkpoint에 함께 담는다.

aligned:    [data][data][data][barrier]  → barrier가 데이터 뒤에서 기다림
unaligned:  [barrier][data][data][data]  → barrier가 먼저 가고, 뒤의 data는 함께 저장

결국 Chandy-Lamport처럼 채널 내용을 저장하는 쪽으로 돌아간 셈이다. backpressure가 심할 때 checkpoint가 끝나지 않는 문제를 풀려고 저장 I/O를 더 쓰는 선택이다.

5장에서 볼 buffer debloating도 같은 문제를 다른 방향에서 푼다. barrier가 앞지르게 하는 대신, buffer에 쌓아 두는 데이터 자체를 줄여 barrier가 덜 기다리게 한다.