이번 시간에는 Flink SQL의 GROUP BY 집계에 State TTL을 걸어 보면서, 체크포인트 크기가 어떻게 달라지는지 Docker로 띄운 Flink와 Kafka에서 측정하며 배워가는 시간입니다.
예를 들어서 우리가 사용자별 클릭 수를 세는 Flink SQL 잡을 오래 띄워 둔다고 해볼게요. 그러면 체크포인트가 점점 커지고 TaskManager 메모리 사용량도 따라 오릅니다.
GROUP BY user_id 집계는 사용자마다 상태를 하나씩 들고 있고, 한 번 만든 상태는 그 사용자가 다시 오지 않아도 지우지 않기 때문이에요. 이걸 줄이는 설정이 State TTL입니다.
실습 준비
이 글의 명령은 모두 실습 코드의 flink-state-ttl/ 디렉터리에서 실행합니다.
compose.yaml: Kafka 브로커, Flink JobManager, TaskManager 세 컨테이너 정의common.sh:kafka,reset_topic,feed_keys,cancel_all,submit,ckpt_watch등 셸 함수sql/: 제출하는 Flink SQL 잡 두 개(count.sql,count-ttl.sql)
그 밖의 파일은 데이터·보조 스크립트입니다.
구성은 Kafka 브로커 하나와 Flink JobManager, TaskManager 하나씩이에요. Flink 잡은 Flink SQL로 만들고 공식 이미지의 SQL 클라이언트(sql-client.sh)로 제출합니다.
Kafka SQL 커넥터 jar는 아래 명령으로 lib/에 받아 두면 돼요. compose.yaml이 이 jar를 두 Flink 컨테이너의 /opt/flink/lib에 마운트합니다.
mkdir -p lib sql ckpt
curl -sSfL -o lib/flink-sql-connector-kafka-5.0.0-2.2.jar \
https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/5.0.0-2.2/flink-sql-connector-kafka-5.0.0-2.2.jar
ls lib
flink-sql-connector-kafka-5.0.0-2.2.jar
compose.yaml은 Kafka 브로커, JobManager, TaskManager 세 컨테이너를 띄웁니다. 설정 중 이 글에서 중요한 것은 이렇습니다.
taskmanager.memory.managed.fraction: ${MANAGED_FRACTION:-0.1}: managed memory는 Flink가 JVM 힙 밖에 따로 떼어 두는 메모리로, RocksDB 상태 백엔드가 써요. 이 글은 RocksDB를 쓰지 않아서 0.1로 낮췄고, 값을 바꾸는MANAGED_FRACTION변수도 이 글에서는 쓰지 않아요.taskmanager.numberOfTaskSlots: 1: 이 실습은 한 번에 잡 하나만 돌립니다.execution.checkpointing.dir: file:///ckpt: 체크포인트를 저장할 곳이에요. 실습 디렉터리의ckpt를 두 Flink 컨테이너의/ckpt에 bind mount했습니다. macOS Docker Desktop에서는 이미지 안의flink사용자(uid 9999)가 권한 설정 없이 이 디렉터리에 썼어요.execution.checkpointing.interval: 10 s: 10초마다 체크포인트를 만듭니다.execution.checkpointing.incremental: false: 증분 체크포인트는 바뀐 부분만 올리고 크기도 그만큼만 보여 줘요. 상태 전체 크기를 비교하려고 증분을 껐습니다. 기본값도 false지만 명시했어요.metrics.fetcher.update-interval: 1000: REST API가 TaskManager에서 지표를 다시 가져오는 간격이에요. 기본 10초를 1초로 줄였습니다.- JobManager의 8081 포트(Web UI와 REST API)는 호스트의 7670으로 엽니다.
compose 프로젝트 이름을 flkst로 정해서 띄운 다음, 클러스터가 슬롯 1개를 인식했는지 확인해 볼게요.
export COMPOSE_PROJECT_NAME=flkst
docker compose up -d
until curl -s localhost:7670/overview | grep -q '"slots-total":1'; do sleep 2; done
curl -s localhost:7670/overview; echo
{"taskmanagers":1,"slots-total":1,"slots-available":1,"jobs-running":0,"jobs-finished":0,"jobs-cancelled":0,"jobs-failed":0,"flink-version":"2.2.1","flink-commit":"450c63e"}
실습에서 반복해서 쓰는 명령은 common.sh에 셸 함수로 모아 두었어요. REST 응답은 jq로 읽습니다. macOS 15 이상에는 /usr/bin/jq가 들어 있어요.
kafka: Kafka 컨테이너 안에서 Kafka CLI를 실행해요. 브로커와 같은 컨테이너에서 돌기 때문에 CLI의 힙을 128MB로 줄였습니다.reset_topic:keys토픽을 지우고 파티션 1개로 다시 만들어요.feed_keys <시작> <끝> <초당 건수>: Kafka 컨테이너 안에서seq로{"user_id":1}같은 JSON을 만들어 1초에 정해진 건수씩 보냅니다. 프로듀서 하나로 끝까지 보내고, 모든 키가 서로 달라요. 끝까지 보내면 입력이 멈춥니다.job_id: REST API에서RUNNING인 잡의 ID를 찾아 출력해요.cancel_all: REST API로 실행 중인 잡을 모두 찾아 취소하고취소: <잡 ID>를 한 줄씩 출력해요.submit: 실행 중인 잡을 취소하고 SQL 파일을 제출한 뒤 잡이RUNNING이 될 때까지 기다립니다.ckpt_watch <초>: 정한 시간 동안 체크포인트가 새로 끝날 때마다 한 줄씩 출력해요. 체크포인트 크기는 REST API/jobs/<잡 ID>/checkpoints의state_size이고, 커밋 offset은 Kafka 소스가 그 체크포인트와 함께 Kafka에 커밋한 위치입니다. 키 하나가 메시지 하나라서 커밋 offset은 그 체크포인트까지 읽은 키 수와 같아요.
sql/count.sql은 keys 토픽을 읽어 user_id별 건수를 셉니다. 결과는 받은 행을 그냥 버리는 blackhole 싱크로 보내서, 상태 크기만 볼 때 결과 출력이 로그나 디스크를 채우지 않아요.
토픽을 만들고 잡을 제출한 뒤, 체크포인트가 몇 번 돌도록 25초를 기다렸다가 체크포인트 응답에서 크기 필드를 확인해 볼게요.
source common.sh
kafka kafka-topics.sh --create --topic keys --partitions 1
submit count.sql
sleep 25
curl -s localhost:7670/jobs/$(job_id)/checkpoints | jq '.latest.completed | {id, status, state_size, checkpointed_size, external_path}'
ls ckpt ckpt/*
Created topic keys.
Job ID: 2829395a82857562f86734aaab1a0626
RUNNING: 2829395a82857562f86734aaab1a0626
{
"id": 3,
"status": "COMPLETED",
"state_size": 5835,
"checkpointed_size": 5835,
"external_path": "file:/ckpt/2829395a82857562f86734aaab1a0626/chk-3"
}
ckpt:
2829395a82857562f86734aaab1a0626
ckpt/2829395a82857562f86734aaab1a0626:
chk-3
shared
taskowned
state_size는 체크포인트 전체 크기이고, checkpointed_size는 이번 체크포인트에서 새로 올린 크기예요. 증분 체크포인트를 끈 상태라 둘이 같습니다.
실습 1: 키 수에 따라 체크포인트가 커집니다
이제 TTL 없이 키 24만 개를 초당 2,000개씩 넣으면서 150초 동안 체크포인트를 지켜볼게요. 토픽을 비우고 잡을 다시 제출한 다음, 키는 백그라운드로 보냅니다.
cancel_all; reset_topic
submit count.sql
feed_keys 1 240000 2000 >/dev/null 2>&1 &
ckpt_watch 150
취소: f6fb0048119fac48ab7269ad3719c85f
Created topic keys.
Job ID: 4daacac638f4efce78ab8090e14f496d
RUNNING: 4daacac638f4efce78ab8090e14f496d
01:02:31 chk-1 커밋 offset 7464 체크포인트 304 KB
01:02:42 chk-2 커밋 offset 27849 체크포인트 1120 KB
01:02:52 chk-3 커밋 offset 47965 체크포인트 1926 KB
01:03:02 chk-4 커밋 offset 67965 체크포인트 2726 KB
01:03:12 chk-5 커밋 offset 87965 체크포인트 3527 KB
01:03:23 chk-6 커밋 offset 107576 체크포인트 4312 KB
01:03:30 chk-7 커밋 offset 107576 체크포인트 5126 KB
01:03:40 chk-8 커밋 offset 147890 체크포인트 5927 KB
01:03:51 chk-9 커밋 offset 167890 체크포인트 6727 KB
01:04:01 chk-10 커밋 offset 187890 체크포인트 7528 KB
01:04:12 chk-11 커밋 offset 205890 체크포인트 8249 KB
01:04:22 chk-12 커밋 offset 225890 체크포인트 9050 KB
01:04:32 chk-13 커밋 offset 225890 체크포인트 9615 KB
01:04:43 chk-14 커밋 offset 240000 체크포인트 9615 KB
01:04:50 chk-15 커밋 offset 240000 체크포인트 9615 KB
체크포인트 크기가 10초마다 약 800KB씩 늘어, 키 24만 개를 다 읽었을 때 9615KB가 됐어요. 입력이 멈춘 뒤(chk-14, chk-15)에는 더 늘지 않았지만 줄지도 않았습니다. 같은 실험을 세 번 했고 세 번 모두 마지막 크기가 9615KB였어요.
커밋 offset은 체크포인트가 끝난 뒤 소스가 Kafka에 따로 커밋하는 값이라, 커밋이 지표에 늦게 반영되면 chk-7과 chk-13처럼 앞 줄과 같은 값이 한 번 더 찍히고 그다음 줄에서 따라잡습니다.
체크포인트 크기는 직렬화 형식과 부가 정보를 포함한 값이라, 이 숫자를 "키 하나에 몇 바이트"로 읽으면 안 됩니다. 실제 서비스에서 새 사용자, 새 세션 ID, 새 주문 번호처럼 키가 계속 새로 생기면 상태는 잡이 도는 내내 커집니다.
실습 2: table.exec.state.ttl을 설정하면
table.exec.state.ttl은 일정 시간 갱신되지 않은 키의 상태를 만료시키는 설정이에요. 기본값은 0이고, 0이면 상태를 지우지 않습니다.
sql/count-ttl.sql은 count.sql 맨 위에 SET 'table.exec.state.ttl' = '30 s'; 한 줄만 더한 파일이에요.
TTL을 30초로 주고 같은 입력을 같은 속도로 보내 볼게요.
cancel_all; reset_topic
submit count-ttl.sql
feed_keys 1 240000 2000 >/dev/null 2>&1 &
ckpt_watch 135
취소: 4daacac638f4efce78ab8090e14f496d
Created topic keys.
Job ID: 7a2950077c787cc6fe79d679fce6958d
RUNNING: 7a2950077c787cc6fe79d679fce6958d
01:05:42 chk-1 커밋 offset -1 체크포인트 5 KB
01:05:53 chk-2 커밋 offset 13412 체크포인트 647 KB
01:06:03 chk-3 커밋 offset 33965 체크포인트 1631 KB
01:06:13 chk-4 커밋 offset 53965 체크포인트 2588 KB
01:06:24 chk-5 커밋 offset 71965 체크포인트 3048 KB
01:06:34 chk-6 커밋 offset 91965 체크포인트 3020 KB
01:06:44 chk-7 커밋 offset 111890 체크포인트 3011 KB
01:06:52 chk-8 커밋 offset 131761 체크포인트 3013 KB
01:07:02 chk-9 커밋 offset 151890 체크포인트 3014 KB
01:07:12 chk-10 커밋 offset 171890 체크포인트 3018 KB
01:07:23 chk-11 커밋 offset 191890 체크포인트 3012 KB
01:07:33 chk-12 커밋 offset 211890 체크포인트 3015 KB
01:07:43 chk-13 커밋 offset 231890 체크포인트 3016 KB
chk-1의 커밋 offset -1은 아직 커밋한 offset이 없다는 뜻이에요.
처음 30초 동안은 TTL이 없을 때처럼 늘다가, chk-5부터는 3011~3048KB 사이에서 더 늘지 않았습니다. 키는 24만 개까지 계속 들어왔지만, 새 키가 들어오는 만큼 30초 넘게 갱신되지 않은 키가 지워졌기 때문이에요. 세 번 실행해서 크기가 멈춘 구간의 값은 3009~3090KB였어요.
TTL은 처리 시간 기준이에요. 이벤트에 적힌 시간이 아니라 Flink가 상태를 갱신한 시간부터 셉니다.
실습 3: 만료와 정리는 다릅니다
앞 실험에서 입력이 멈춘 뒤에도 잡을 그대로 두고 2분(TTL의 4배)을 더 지켜볼게요. 이 시점에 상태에 남은 키는 모두 30초 넘게 갱신되지 않았으니 만료됐어야 합니다. 출력은 처음 세 줄과 마지막 줄만 실었어요(발췌).
ckpt_watch 130
01:07:51 chk-13 커밋 offset 231890 체크포인트 3016 KB
01:07:55 chk-14 커밋 offset 231890 체크포인트 3019 KB
01:08:02 chk-15 커밋 offset 240000 체크포인트 3019 KB
01:09:53 chk-26 커밋 offset 240000 체크포인트 3019 KB
TTL의 4배가 지났는데도 크기는 3019KB 그대로였어요. 세 번 모두 크기가 바뀌지 않았습니다(각각 3014KB, 3019KB, 3040KB). 왜 그럴까요?
설정하지 않으면 Flink는 상태를 TaskManager의 JVM 힙에 두는 HashMapStateBackend(이 글에서는 힙 상태 백엔드)를 씁니다. Flink 2.2 문서에 따르면 힙 상태 백엔드는 만료된 상태를 증분 정리(incremental cleanup)로 지워요. 어떤 키든 상태에 접근할 때마다 만료된 항목을 몇 개씩(기본값은 접근 한 번에 5개) 확인해 지우니까, 입력이 멈추면 정리도 멈추는 거예요.
그럼 이번에는 새 키를 보내서 정리가 다시 일어나는지 볼게요. 남은 키는 대략 6만 개(초당 2,000개 × 30초)이므로, 그 1/100부터 1/5까지(600, 1,200, 3,000, 6,000, 12,000건) 누적 건수가 되도록 새 키를 나눠 보내고 단계마다 체크포인트 두 번을 기다립니다.
n=240000
for total in 600 1200 3000 6000 12000; do
echo "== 새 키 누적 ${total}건"
feed_keys $((n + 1)) $((240000 + total)) 2000 >/dev/null 2>&1
n=$((240000 + total))
ckpt_watch 22
done
== 새 키 누적 600건
01:10:07 chk-27 커밋 offset 240000 체크포인트 3019 KB
01:10:14 chk-28 커밋 offset 240000 체크포인트 2805 KB
01:10:24 chk-29 커밋 offset 240600 체크포인트 2805 KB
== 새 키 누적 1200건
01:10:35 chk-30 커밋 offset 240600 체크포인트 2805 KB
01:10:46 chk-31 커밋 offset 240600 체크포인트 2589 KB
01:10:56 chk-32 커밋 offset 241200 체크포인트 2589 KB
== 새 키 누적 3000건
01:11:06 chk-33 커밋 offset 241200 체크포인트 2589 KB
01:11:16 chk-34 커밋 offset 241200 체크포인트 1954 KB
01:11:27 chk-35 커밋 offset 243000 체크포인트 1954 KB
== 새 키 누적 6000건
01:11:38 chk-36 커밋 offset 243000 체크포인트 1954 KB
01:11:45 chk-37 커밋 offset 243000 체크포인트 911 KB
01:11:55 chk-38 커밋 offset 246000 체크포인트 911 KB
== 새 키 누적 12000건
01:12:09 chk-39 커밋 offset 246000 체크포인트 911 KB
01:12:16 chk-40 커밋 offset 246000 체크포인트 372 KB
01:12:26 chk-41 커밋 offset 252000 체크포인트 372 KB
feed_keys가 키를 다 보낸 뒤에 ckpt_watch가 시작하기 때문에, 각 단계의 첫 줄은 키를 보내기 전에 끝난 체크포인트입니다.
크기는 새 키를 처리한 뒤에 줄었어요. 새 키 600건을 처리하자 3019KB가 2805KB로 줄었고, 보낸 건수가 늘수록 크기가 단계적으로 줄어 12,000건 뒤에는 372KB가 됐습니다. 세 번 실행에서 12,000건 뒤의 크기는 372~457KB였어요.
600건에 6만 개 가까운 항목이 지워지지 않은 건 정리가 한 번에 몇 개씩만 일어나기 때문이에요.
정리
범위: Flink 2.2.1(flink:2.2.1-java17), Kafka 4.3.1(apache/kafka:4.3.1), TaskManager 1개(슬롯 1개), 새 키 24만 개를 초당 2,000개, 비증분 체크포인트 10초 간격, 주요 실험은 3번 반복.
- TTL 없이 키 24만 개를 넣자 체크포인트가 10초마다 약 800KB씩 늘어 9615KB가 됐고, 입력이 멈춘 뒤에도 줄지 않았습니다. 새 사용자나 새 주문 번호처럼 키가 계속 새로 생기는
GROUP BY집계는 상태가 잡이 도는 내내 커진다고 보고, 체크포인트 크기를 키 수 증가와 함께 추세로 보세요. table.exec.state.ttl을 30초로 주자 체크포인트가 3009~3090KB에서 더 늘지 않았습니다. 상태 크기는 TTL 동안 들어오는 고유 키 수로 가늠해 보세요.- 힙 상태 백엔드에서 입력이 멈춘 뒤 TTL의 4배인 2분이 지나도 크기는 3019KB 그대로였고, 새 키를 600건에서 12,000건까지 보내자 2805KB에서 372KB로 단계적으로 줄었습니다. 입력이 끊기는 시간대에는 만료된 상태가 정리되지 않는다고 보고 크기를 잡으세요.
- 이 글에 싣지 않은 추가 측정에서는 TTL 15초인 같은 집계에 키 1을 두 번 보내고 50초 뒤 다시 보내자, 건수가 3이 아니라 1로 처음부터 셌습니다. 업무 규칙에 기간이 없는 누적 집계에는 크기를 줄이려고 TTL을 걸지 말고, TTL은 같은 키가 다시 오는 가장 긴 간격보다 길게 잡으세요.