2. State

checkpoint가 상태를 언제, 어떻게 일관되게 찍는지의 문제라면, 이 장에서는 그 상태가 무엇이고, 어디에 있고, 병렬도를 바꿀 때 어떻게 옮겨지는지를 다룬다. 2017년 논문의 핵심 기여는 상태를 사용자 코드가 아니라 엔진이 관리하는 것, 즉 managed state로 만든 데 있다.

2.1 Why the Engine Must Know the State

사용자별 최근 30분 클릭 수를 계산하려면 operator가 사용자마다 숫자를 들고 있어야 한다. 이 숫자를 operator 안의 평범한 HashMap에 두면 엔진은 그 안에 무엇이 있는지 모른다. 그러면 엔진은 스냅샷을 찍는 일도, 장애를 복구하는 일도, 병렬도를 바꾸는 일도 대신해 줄 수 없다.

그래서 Flink는 상태를 엔진이 제공하는 API로 선언하게 한다. 스트림 job은 오래 실행되므로 중간 계산 상태도 계속 유지해야 한다. 그리고 엔진이 상태의 구조를 알아야 그 상태를 직렬화해 스냅샷을 찍고, 다른 머신으로 옮기고, key 단위로 나눌 수 있다.

2.2 Keyed State and Operator State

managed state는 무엇에 귀속되느냐에 따라 두 가지로 나뉜다.

keyed state는 keyBy 뒤의 operator에서 key마다 따로 있는 상태다. SQL의 GROUP BY처럼 key마다 독립적으로 계산된다. ValueState, ListState, MapState, ReducingState, AggregatingState 같은 형태로 쓰고, 레코드를 처리할 때는 그 레코드의 key에 해당하는 상태만 보인다.

operator state는 key와 상관없이 병렬 subtask 하나에 붙은 상태다. 대표적인 예가 Kafka source의 partition별 offset이다. 4장에서 보겠지만 Flink가 Kafka를 어디까지 읽었는지는 바로 이 operator state에 들어 있다.

flowchart TD
  job["Flink job"] --> s0["subtask 0"]
  job --> s1["subtask 1"]
  s0 --> k0["keyed state<br/>key별 값"]
  s0 --> o0["operator state<br/>partition offset 등"]
  s1 --> k1["keyed state"]
  s1 --> o1["operator state"]

2.3 Key Groups: The Unit of Redistribution

병렬도 4로 돌던 job을 8로 늘린다고 하자. keyed state를 key마다 새 subtask로 옮겨야 하는데, key가 수억 개라면 key 단위로 옮기는 일은 비싸다. 반대로 subtask 하나의 상태 전체를 한 단위로 옮기면 더 잘게 나눠 배분할 수 없다.

Flink는 둘 사이에 key group이라는 중간 단위를 둔다.

flowchart LR
  key["key"] -->|"hash mod max parallelism"| kg["key group"]
  kg -->|"연속된 범위를 배정"| st["subtask"]

1. key를 해시해 고정된 개수의 key group 중 하나에 넣는다. 이 개수가 job의 최대 병렬도(max parallelism)다. 따로 정하지 않으면 병렬도를 보고 자동으로 정하는데, 최소 128이다.
2. 각 subtask는 연속된 key group 범위를 맡는다. 최대 병렬도가 128이고 병렬도가 4면 subtask마다 32개씩 맡는다.
3. 병렬도를 바꾸면 key group 범위만 다시 나눈다. 상태 파일도 key group 단위로 쓰여 있으므로, 새 subtask는 자기 범위의 key group 상태만 읽어 오면 된다.

결과적으로 재분배 비용은 key 개수와 상관없어지고, key group 개수만큼의 묶음을 옮기는 문제가 된다. 논문은 이를 재구성할 때의 I/O와 스냅샷 메타데이터 크기 사이의 절충으로 설명한다. key group이 너무 적으면 늘릴 수 있는 병렬도에 한계가 생기고, 너무 많으면 관리할 메타데이터가 늘어난다.

대가도 있다. 최대 병렬도는 처음 정하면 바꾸기 어렵다. key group 개수가 상태 파일의 구조에 들어가기 때문이다. 나중에 바꾸면 기존 상태와 호환되지 않는다. 그래서 크게 늘릴 가능성이 있는 job은 처음부터 pipeline.max-parallelism을 넉넉히 잡는다.

이 구조는 데이터 쏠림과도 이어진다. Kafka partition key나 keyBy key가 한쪽으로 쏠리면, 특정 key group과 그 key group을 맡은 subtask에 상태와 처리가 함께 몰린다.

operator state는 key group 없이 나눈다. 병렬도가 바뀌면 각 subtask의 상태 목록을 모아서 균등하게 다시 나누거나(even-split), 모든 subtask에 전체를 나눠 주는(union) 방식을 쓴다.

2.4 State Backends: Where State Lives

상태에 접근하는 방법과 스냅샷을 찍는 방법은 state backend가 정한다. checkpoint 알고리즘은 언제 찍을지만 정하고, 어떻게 찍을지는 backend가 맡는다. backend는 state.backend.type으로 고른다.

여기서 헷갈리기 쉬운 구분이 하나 있다. checkpoint 파일 자체는 어느 backend든 S3 같은 내구 저장소에 쓴다. backend는 실행 중인 상태를 어디에 둘지 정하고, checkpoint 저장소는 스냅샷을 어디에 남길지 정한다. HashMap backend를 써도 checkpoint는 S3에 쌓인다.

2.5 Incremental Checkpoints

RocksDB를 쓰는 큰 이유가 증분 checkpoint다. RocksDB는 LSM tree라서 데이터를 불변 SST 파일로 쓴다. checkpoint마다 전체 상태를 다시 올리는 대신, 지난 checkpoint 이후 새로 생긴 SST 파일만 올리고 나머지는 참조로 공유한다. 상태가 수백 GB여도 checkpoint 한 번에 올리는 양은 변경분 정도로 유지된다.

증분 checkpoint는 execution.checkpointing.incremental로 켠다. 켜면 UI에 보이는 checkpoint 크기도 변경분만 표시되므로, 전체 상태 크기와 헷갈리지 않도록 주의한다.

비동기 스냅샷도 backend가 맡는다. 상태를 쓰는 동안 처리를 멈추지 않으려면, backend가 현재 상태를 빠르게 고정해 두고(copy-on-write나 RocksDB snapshot) 실제 업로드는 background thread에서 해야 한다.

2.6 Keeping State from Growing Forever

사용자 id나 세션 id처럼 key가 계속 새로 생기는 job에서는 상태가 끝없이 늘어난다. keyed state에 TTL을 걸면 일정 시간 갱신되지 않은 key의 상태를 지운다. window operator는 watermark가 window 끝을 지나면 그 window의 상태를 알아서 정리한다. 페이지나 세션 단위로 집계하는 job이라면 처음부터 TTL을 설계에 넣어 두는 편이 좋다.