6편 · 약 27분

Flink 상태 운영: state backend, checkpoint, savepoint, rescaling

Flink 운영의 진짜 난점은 “계산”보다 “상태”다

Flink 잡이 단순히 Kafka에서 읽어서 필터링한 뒤 다른 topic으로 내보내는 정도라면 운영 난이도는 비교적 낮다. 하지만 실무의 스트리밍 잡은 대개 상태를 가진다. 최근 10분 윈도우 집계, 사용자별 마지막 이벤트, 주문별 중복 제거 키, join을 위한 임시 버퍼, 외부 sink transaction 정보처럼 이전 이벤트를 기억해야 다음 이벤트를 올바르게 처리할 수 있는 정보가 생긴다.

상태가 생기면 장애 대응 방식도 달라진다. 프로세스를 다시 띄우는 것만으로는 충분하지 않다. 어느 시점의 state와 어느 Kafka offset을 함께 복구할지, checkpoint 저장소가 실제로 durable한지, savepoint로 업그레이드할 수 있는지, parallelism을 바꿀 때 key가 어느 subtask로 재분배되는지까지 같이 봐야 한다.

이 장에서는 Flink 상태 운영을 네 가지 질문으로 정리한다.

state backend는 어디에 working state를 둘 것인가?
checkpoint는 장애 후 어디까지 자동 복구해 줄 것인가?
savepoint는 배포·업그레이드·마이그레이션을 어떻게 안전하게 만들 것인가?
rescaling은 상태를 어떤 단위로 나누고 다시 배치하는가?

1. Flink state를 운영 관점에서 다시 보기

Flink state는 크게 두 층으로 나누어 생각하면 쉽다.

구분	의미	운영자가 보는 지표/위험
Working state	TaskManager가 현재 이벤트 처리에 사용하는 로컬 상태	메모리, local disk, RocksDB latency, GC, serialization 비용
State snapshot	checkpoint/savepoint로 durable storage에 저장한 복구 지점	checkpoint duration, alignment time, failed checkpoint, storage 비용, restore time

중요한 점은 state backend가 “복구 파일을 어디에 둘 것인가”만 정하는 설정이 아니라는 것이다. backend는 working state의 표현 방식과 접근 비용에도 영향을 준다. checkpoint storage는 snapshot을 어디에 저장할지에 더 가깝다. 둘을 섞어 이해하면 “RocksDB를 쓰니까 자동으로 S3에 안전하게 저장되겠지” 같은 위험한 착각이 생긴다.

Kafka / Source

→

Flink operator
keyed/window state

→

State backend
heap 또는 RocksDB

→

Sink
transaction / upsert

Checkpoint barrier가 stream position과 state를 같은 복구 지점으로 묶는다

↓

Checkpoint storage: HDFS/S3/object storage 등 모든 TaskManager가 접근 가능한 durable 경로

↓

장애 복구: 마지막 완료 checkpoint에서 state와 source position을 함께 복원

Flink state 운영 경로

운영자가 잡을 볼 때는 단순히 “checkpoint가 켜져 있는가?”보다 다음 질문을 먼저 해야 한다.

state 크기가 JVM heap에 안정적으로 들어가는가?
checkpoint 경로가 JobManager 로컬 디스크가 아니라 durable shared storage인가?
checkpoint duration이 interval보다 길어져 checkpoint backlog가 생기지 않는가?
실패 시 restore time이 RTO 안에 들어오는가?
배포 전 savepoint를 만들고, 그 savepoint로 실제 restore 리허설을 해 봤는가?

2. State backend 선택: HashMap과 RocksDB의 tradeoff

Flink의 stable 문서 기준으로 기본 state backend는 HashMapStateBackend다. 이 backend는 state를 JVM heap의 Java object로 보관한다. 접근 속도가 빠르고 구현을 이해하기 쉽지만, state 크기가 JVM heap과 GC 한계에 묶인다. 작은 state, 낮은 latency가 중요한 job에는 적합할 수 있다.

반대로 EmbeddedRocksDBStateBackend는 working state를 TaskManager 안에 embedded된 RocksDB에 직렬화된 key/value byte 형태로 저장한다. RocksDB는 별도 클러스터가 아니라 TaskManager 프로세스 안에서 JNI/native thread와 local file을 사용하는 embedded key-value store다. 장점은 state가 JVM heap보다 훨씬 커질 수 있고, incremental checkpoint를 사용할 수 있다는 점이다. 단점은 모든 state read/write에 serialization/deserialization과 RocksDB 접근 비용이 붙기 때문에 heap 기반보다 느릴 수 있다는 점이다.

선택지	장점	비용/주의점	적합한 상황
`HashMapStateBackend`	heap object 접근이라 빠름, 단순함	heap/GC 영향, 큰 state에 취약	작은 keyed state, low-latency 처리, 개발·검증
`EmbeddedRocksDBStateBackend`	큰 state, local disk 활용, incremental checkpoint	serialization/JNI/disk 비용, RocksDB memory·disk 튜닝 필요	long window, 대규모 keyed state, state 크기 증가가 예상되는 운영 job

설정 예시는 다음처럼 읽으면 된다. 실제 key 이름은 Flink 버전에 따라 조금씩 달라질 수 있으므로 운영 환경의 버전 문서를 확인해야 한다.

# flink-conf.yaml 예시
state.backend.type: rocksdb
execution.checkpointing.dir: s3://company-flink-checkpoints/prod/order-aggregation
execution.checkpointing.interval: 60s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.incremental: true

여기서 state.backend.type은 working state의 backend를, execution.checkpointing.dir은 checkpoint metadata와 state 파일을 저장할 durable 경로를 뜻한다. 운영에서 가장 흔한 실수는 checkpoint directory를 local path로 두거나, 개발용 JobManagerCheckpointStorage에 가까운 구성을 production에 가져가는 것이다. JobManager heap에 checkpoint snapshot을 두는 방식은 local development나 매우 작은 state에는 편하지만, HA 운영과 큰 state에는 맞지 않는다.

3. Checkpoint: 자동 장애 복구의 기준점

checkpoint는 Flink가 자동으로 만드는 복구 지점이다. checkpoint가 완료되면 Flink는 state와 source position을 함께 기억한다. 장애가 나면 마지막으로 완료된 checkpoint에서 state를 복원하고, source도 그 지점부터 다시 읽어서 장애가 없었던 것과 같은 처리 의미론에 가깝게 돌아간다.

운영자는 checkpoint를 세 가지 숫자로 봐야 한다.

지표	봐야 하는 이유	나쁜 신호
`checkpoint duration`	snapshot이 끝나는 데 걸리는 시간	interval보다 길어짐, 계속 증가
`checkpoint size`	state 증가와 저장소 비용을 반영	급증, incremental delta가 예상보다 큼
`failed checkpoints`	복구 가능성 자체의 신뢰도	timeout, storage error, barrier alignment 지연 반복

checkpoint interval은 “짧을수록 무조건 좋다”가 아니다. 짧게 잡으면 장애 후 재처리 범위가 줄고 transactional sink commit이 더 자주 일어날 수 있다. 하지만 snapshot overhead, storage write, network I/O가 늘어난다. 반대로 너무 길면 장애 후 reprocessing 범위가 커지고 restore 후 catch-up 시간이 길어진다.

실무에서는 다음 순서로 조정한다.

RPO/RTO 요구를 먼저 정한다. 예: 장애 후 최대 2분 재처리 허용, 15분 내 복구.
checkpoint interval을 정한다. 예: 30초, 60초, 5분.
실제 state 크기와 throughput에서 checkpoint duration이 interval보다 충분히 짧은지 본다.
checkpoint timeout, min pause, max concurrent checkpoint를 함께 조정한다.
checkpoint storage의 쓰기 throughput과 object storage API throttling을 확인한다.

RocksDB incremental checkpoint를 켜면 매번 전체 state를 쓰지 않고 이전 완료 checkpoint 이후 변경분 중심으로 snapshot을 만든다. 큰 state job에서는 checkpoint 시간을 크게 줄일 수 있다. 다만 restore 때 여러 delta를 가져와야 하므로 병목이 network인지 CPU/IOPS인지에 따라 restore 시간이 달라질 수 있다. incremental checkpoint UI의 “checkpointed data size”도 전체 state 크기가 아니라 해당 checkpoint의 delta 크기로 해석해야 한다.

4. Savepoint: 배포·업그레이드·마이그레이션용 수동 스냅샷

checkpoint와 savepoint는 둘 다 state snapshot이지만 목적이 다르다.

구분	Checkpoint	Savepoint
주 목적	예기치 못한 장애 자동 복구	계획된 운영 작업: 배포, 업그레이드, 마이그레이션, fork
생성 주체	Flink가 주기적으로 관리	사용자가 명시적으로 trigger
lifecycle	Flink가 소유·정리하는 성격이 강함	사용자가 보관·삭제 책임을 갖는 경우가 많음
최적화 방향	빠른 생성과 빠른 복구	이식성, 운영 유연성

운영 배포에서 savepoint가 중요한 이유는 “코드를 바꿔도 state를 잃지 않기 위해서”다. 예를 들어 주문 집계 job의 window state가 수백 GB인데 배포 때마다 새로 집계를 시작하면 데이터가 깨진다. savepoint를 만들고, 새 job을 그 savepoint에서 시작하면 기존 state를 이어받을 수 있다.

하지만 savepoint restore는 operator state mapping에 의존한다. Flink 문서는 DataStream job의 stateful operator에 명시적인 uid(String)를 부여할 것을 강하게 권장한다. uid를 주지 않으면 Flink가 자동 생성한 ID에 의존하게 되는데, operator 순서 변경이나 중간 연산 추가처럼 코드 구조가 조금만 바뀌어도 ID가 달라져 restore가 실패하거나 의도치 않은 state mapping 문제가 생길 수 있다.

배포 전 체크리스트는 다음과 같다.

stateful operator에 안정적인 uid가 있는가?
새 코드가 기존 serializer와 schema evolution을 지원하는가?
savepoint를 만든 뒤 새 job을 staging에서 restore해 봤는가?
transactional sink state를 이어받아도 안전한가?
restore 후 첫 checkpoint가 성공하기 전까지 원본 savepoint를 삭제하지 않는가?
실패 시 이전 artifact와 savepoint로 되돌아가는 절차가 문서화되어 있는가?

예시 운영 절차는 다음처럼 잡는다.

# 1. 실행 중인 job의 savepoint 생성
bin/flink savepoint <jobId> s3://company-flink-savepoints/order-aggregation/

# 2. 새 artifact를 savepoint에서 시작
bin/flink run \
  -s s3://company-flink-savepoints/order-aggregation/savepoint-xxxx \
  -p 24 \
  order-aggregation-new.jar

# 3. 새 job의 첫 checkpoint 성공 확인 후 이전 job/savepoint 정리

명령 자체보다 중요한 것은 “첫 checkpoint 성공 전에는 복구 기준점이 아직 savepoint에 강하게 의존한다”는 운영 감각이다. restore 직후 장애가 나면 Flink가 초기 snapshot에서 다시 복구하려 할 수 있으므로, savepoint 파일을 너무 빨리 지우면 안 된다.

5. Rescaling: parallelism 변경은 state 재배치다

Flink job의 parallelism을 바꾸는 것은 단순히 worker 수를 늘리는 일이 아니다. keyed state가 있는 job에서는 key별 상태가 subtask 사이에 다시 배치된다. Flink는 key를 key-group이라는 단위로 나누고, 이 key-group을 parallel subtask에 할당한다. max parallelism은 이 key-group 수의 상한으로 이해할 수 있다.

그래서 savepoint를 사용할 계획이 있으면 max parallelism을 초기에 신중하게 잡아야 한다. parallelism은 나중에 바꿀 수 있지만, max parallelism은 state partitioning의 상한과 관련되어 있어 운영 중 변경이 훨씬 민감하다. 너무 낮게 잡으면 나중에 scale-out 한계가 되고, 너무 높게 잡으면 관리해야 할 key-group이 늘어 overhead가 생길 수 있다.

parallelism 2

subtask 0
key-group 0,1,2,3

subtask 1
key-group 4,5,6,7

savepoint 또는 retained checkpoint에서 restore하며 parallelism 변경

parallelism 4

subtask 0
0,1

subtask 1
2,3

subtask 2
4,5

subtask 3
6,7

Rescaling과 key-group 재배치

운영에서 rescaling을 계획할 때는 다음을 확인한다.

확인 항목	이유
source partition 수	Kafka partition 수보다 source parallelism을 크게 잡아도 모든 subtask가 균등하게 바쁘지 않을 수 있음
keyed state 크기	restore와 redistribution 시간이 state 크기에 비례해 늘어남
`max parallelism`	미래 scale-out 상한과 key-group 수에 영향
checkpoint/savepoint 성공 여부	rescale restore의 출발점이 필요함
sink 동작	parallelism 변경 후 connection 수, transaction 수, file writer 수가 늘어날 수 있음

Flink의 reactive mode처럼 리소스 변화에 맞춰 자동으로 parallelism을 조정하는 방식도 있지만, 이것 역시 checkpoint에서 restore하는 재시작을 동반한다. 따라서 “autoscaling을 켰다”는 말은 “checkpoint가 안정적이고 restore 시간이 운영 SLO 안에 있다”는 전제를 포함해야 한다.

6. 장애 시나리오별 판단법

6.1 Checkpoint가 계속 실패한다

먼저 실패 원인을 storage, state size, alignment/backpressure로 나눠 본다.

storage: S3/HDFS 권한, DNS, throttling, disk full, object lifecycle 정책을 확인한다.
state size: 특정 operator의 state가 급증했는지, window cleanup이 정상인지 본다.
backpressure/alignment: upstream/downstream 병목 때문에 barrier가 늦게 모이는지 본다.

checkpoint 실패가 오래 지속되면 “지금 job은 장애가 나면 어디까지 복구 가능한가?”를 다시 물어야 한다. 마지막 완료 checkpoint가 너무 오래됐다면 장애 시 재처리 범위가 커지고, external sink와의 exactly-once 경계도 약해질 수 있다.

6.2 Restore가 너무 느리다

restore time은 state 크기, checkpoint 파일 수, object storage read throughput, TaskManager local disk, RocksDB rebuild 비용에 영향을 받는다. incremental checkpoint는 checkpoint 시간을 줄이지만 restore에서 delta 파일을 많이 읽을 수 있다. 반대로 full checkpoint/savepoint는 더 큰 snapshot을 만들지만 복구 경로가 단순할 수 있다.

대응은 다음 순서가 안전하다.

restore가 어느 단계에서 느린지 로그와 metrics로 분해한다.
checkpoint storage read throughput과 request throttling을 확인한다.
RocksDB local disk와 managed memory 설정을 점검한다.
필요하면 state TTL, window cleanup, key cardinality 설계를 재검토한다.
RTO가 중요하면 정기 restore drill로 실제 시간을 기록한다.

6.3 배포 후 savepoint restore가 실패한다

대부분은 operator uid, serializer/schema 변경, job graph 변경 문제다. 자동 생성 uid에 의존한 job이라면 작은 코드 변경도 restore mapping을 깨뜨릴 수 있다. stateful operator uid를 안정적으로 고정하고, 호환되지 않는 state 변경은 별도 migration job이나 State Processor API 같은 절차로 다뤄야 한다.

7. 운영 런북: 상태 있는 Flink job 배포 순서

상태 있는 Flink job은 다음 순서를 기본으로 삼는 것이 좋다.

사전 점검

- 최근 checkpoint 성공 시각, duration, size, failed count 확인 - backpressure와 lag가 이미 높은 상태인지 확인 - savepoint 경로와 권한 확인

savepoint 생성

- running job에서 savepoint trigger - savepoint metadata와 state 파일이 durable storage에 있는지 확인

새 버전 restore

- 새 artifact를 savepoint에서 시작 - 필요한 경우 -p로 parallelism 조정 - stateful operator uid mapping 오류가 없는지 확인

안정화 확인

- 첫 checkpoint 성공 확인 - source lag 감소, sink error, throughput, checkpoint duration 확인

정리

- 이전 job 중지 여부 확인 - 보존 정책에 맞춰 오래된 savepoint/checkpoint 삭제 - 배포 기록에 savepoint path, job id, artifact version, rollback 기준점 기록

이 절차의 핵심은 “배포 성공”을 job submission 성공으로 보지 않는 것이다. 상태 있는 스트리밍 job의 배포 완료 기준은 보통 새 버전이 state를 복원했고, 첫 checkpoint를 성공했고, source lag와 sink 오류가 정상 범위로 돌아온 시점이다.

8. DBA·데이터 플랫폼 엔지니어 관점의 체크리스트

DBA나 데이터 플랫폼 엔지니어에게 Flink state 운영은 데이터베이스 운영과 닮아 있다. checkpoint는 recovery log에 가깝고, savepoint는 계획된 백업/마이그레이션 지점에 가깝다. state backend는 storage engine 선택과 비슷한 tradeoff를 만든다.

운영 체크리스트는 다음처럼 정리할 수 있다.

내구성: checkpoint/savepoint가 local disk가 아닌 durable shared storage에 있는가?
복구성: 마지막 완료 checkpoint와 restore drill 시간이 SLO 안에 있는가?
확장성: state 크기 증가와 max parallelism이 향후 트래픽을 감당하는가?
배포 안전성: stateful operator uid, serializer 호환성, rollback savepoint가 준비되어 있는가?
비용: checkpoint interval, retention, incremental snapshot이 object storage 비용을 폭증시키지 않는가?
관측성: checkpoint duration/size/failure, backpressure, busy time, RocksDB metrics, source lag를 함께 보는가?

Flink state 운영을 잘한다는 것은 특정 backend 하나를 외우는 것이 아니다. state 크기, latency, 복구 시간, 배포 빈도, 저장소 비용 사이의 균형을 잡고, 장애가 났을 때 “어느 snapshot에서 어떤 source position으로 돌아갈지”를 설명할 수 있는 상태를 만드는 일이다.

References

Apache Flink Documentation, “State Backends”, https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/state_backends/
Apache Flink Documentation, “Checkpoints”, https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/checkpoints/
Apache Flink Documentation, “Savepoints”, https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/
Apache Flink Documentation, “Checkpoints vs. Savepoints”, https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/checkpoints_vs_savepoints/
Apache Flink Documentation, “Parallel Execution”, https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/execution/parallel/
Apache Flink Blog, Jun Qin, “Using RocksDB State Backend in Apache Flink: When and How”, 2021-01-18, https://flink.apache.org/2021/01/18/using-rocksdb-state-backend-in-apache-flink-when-and-how/
Confluent Documentation, “Configure Checkpointing”, https://docs.confluent.io/cp-flink/current/jobs/configure/checkpointing.html