이번 시간에는 Docker로 띄운 Flink와 Kafka에 순서가 섞인 클릭 이벤트를 보내 보면서, 워터마크가 이벤트 시간 윈도우의 집계를 어떻게 정하는지 측정하며 배워가는 시간입니다.
예를 들어서 우리가 클릭 이벤트를 Kafka로 받아서 Flink로 10초마다 건수를 세는 집계를 만들었다고 해볼게요. 순서가 섞여서 늦게 도착한 이벤트가 있으면, 잡은 오류 없이 돌고 로그에도 경고가 없는데 윈도우 건수가 모자랄 수 있어요.
왜 이런 일이 생길까요? Flink의 이벤트 시간 윈도우는 워터마크("이 시간 이전의 이벤트는 다 왔다고 본다"는 표시)가 윈도우 끝을 지나야 결과를 내고, 그 뒤에 도착한 이벤트는 기본적으로 집계에서 빼기 때문이에요.
실습의 이벤트 시간은 실제 시간이 아니라 프로듀서가 2026-10-01 10:00:00부터 정해서 넣은 값이에요.
실습 준비
이 글의 명령은 모두 실습 코드의 flink-watermark/ 디렉터리에서 실행합니다.
compose.yaml: Kafka 브로커와 Flink JobManager·TaskManagerproduce.py: 이벤트 파일의 초를 JSON 이벤트로 바꿔 지정한 파티션에 0.3초 간격으로 보내는 프로듀서common.sh: 실습에서 반복해서 쓰는 셸 함수 4개wm.py: REST API에서 실행 중인 잡의 워터마크 지표를 골라 시간으로 바꾸는 스크립트sql/: 실습마다 제출하는 Flink SQL 파일 3개(s1.sql~s3.sql)before.txt,after.txt,events.txt:produce.py가 읽는 이벤트 파일
그 밖의 파일은 데이터·보조 스크립트입니다.
Flink 이미지에는 Kafka 커넥터가 들어 있지 않아요. 그래서 Maven Central에서 Flink 2.2용 Kafka SQL 커넥터를 받아 두 컨테이너의 /opt/flink/lib/에 넣어 줄게요.
mkdir -p lib sql
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 브로커(kafka), jobmanager, taskmanager 서비스를 띄웁니다.
taskmanager.numberOfTaskSlots: 2: TaskManager 하나가 동시에 돌릴 수 있는 슬롯 수예요. 이 실습은 한 번에 잡 하나만 돌리니까 2개면 충분합니다.- Kafka의
EXTERNAL리스너는 호스트의 Python 프로듀서가localhost:7251로 접속하는 용도이고, Flink는 컨테이너 네트워크 안에서kafka:9092로 접속해요. - JobManager의 8081 포트(Web UI와 REST API)는 호스트의 7250으로 엽니다.
compose 프로젝트 이름을 flink로 정해서 띄운 다음, 클러스터가 슬롯 2개를 인식했는지 REST API로 확인하고 파티션 1개짜리 토픽 clicks1을 만들어 볼게요.
export COMPOSE_PROJECT_NAME=flink
docker compose up -d
curl -s localhost:7250/overview; echo
docker compose exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic clicks1 --partitions 1
{"taskmanagers":1,"slots-total":2,"slots-available":2,"jobs-running":0,"jobs-finished":0,"jobs-cancelled":0,"jobs-failed":0,"flink-version":"2.2.1","flink-commit":"450c63e"}
Created topic clicks1.
이벤트는 호스트의 Python에서 confluent-kafka 2.15.1로 보내요. produce.py는 이벤트 파일에서 한 줄에 하나씩 초를 읽어 {"user_id":"u2","ts":"2026-10-01 10:00:03.000","seq":7} 같은 JSON을 만들고, 지정한 파티션으로 0.3초 간격으로 보냅니다.
보내기 전에 윈도우별 정답 건수를 먼저 출력하고, 앞에서 보낸 가장 큰 시간보다 작은 이벤트에는 몇 초 늦었는지 표시해 줘요.
0.3초 간격은 Flink가 워터마크를 만드는 기본 주기 200ms(pipeline.auto-watermark-interval)보다 길게 잡은 값이에요. 이벤트를 한꺼번에 보내면 워터마크가 만들어지기 전에 여러 건이 처리되어, 실행마다 결과가 달라질 수 있습니다.
python3 -m venv venv
venv/bin/pip install -q confluent-kafka==2.15.1
source venv/bin/activate
실습에서 반복해서 쓰는 명령은 common.sh에 셸 함수로 모아 두었어요. 각 함수가 하는 일은 다음과 같습니다.
submit: 실행 중인 잡을 취소하고 SQL 파일을 SQL 클라이언트로 제출해요. 그다음 TaskManager 로그에 Kafka 소스가 파티션을 배정받았다는Assigned to partition줄이 나올 때까지 기다립니다. 실습 테이블은 토픽의 마지막 위치부터 읽게(latest-offset) 만들었기 때문에, 소스가 준비되기 전에 보낸 이벤트는 읽지 않아요.watch_results: Flink의print싱크는 결과를 TaskManager의 표준 출력에 찍어요. 이 함수는docker compose logs -f로 TaskManager 출력을 따라가면서 결과 줄만 보여 줍니다.send:produce.py를 실행해요.wm: 실행 중인 잡의 워터마크를 REST API로 읽어 시간으로 바꿔 보여 줍니다. Web UI(http://localhost:7250)의 잡 화면에서 Watermarks 탭을 열어도 같은 값을 볼 수 있어요.
source common.sh
실습 1: 워터마크가 윈도우를 닫습니다
sql/s1.sql은 clicks1을 읽는 clicks 테이블과 결과를 찍는 window_counts 테이블을 만들고, TUMBLE(TABLE clicks, DESCRIPTOR(ts), INTERVAL '10' SECOND)로 10초 텀블링 윈도우를 만들어 건수를 세는 잡을 제출해요. 텀블링 윈도우는 [00~10), [10~20)처럼 겹치지 않고 이어지는 고정 크기 윈도우라고 보시면 됩니다.
clicks 테이블의 WATERMARK FOR ts AS ts - INTERVAL '2' SECOND가 워터마크 정의예요. Flink 2.2 CREATE 문 문서는 이 형태를 "지금까지 본 가장 큰 타임스탬프에서 지정한 지연값을 뺀 워터마크를 낸다"고 설명합니다.
이 글에서는 빼는 값을 지연값이라고 부를게요. 지연값 2초는 가장 큰 이벤트 시간보다 2초까지 늦게 오는 이벤트는 기다리겠다는 뜻이에요. 윈도우 집계는 워터마크가 윈도우 끝을 지나면 그 윈도우의 결과를 한 번 내고 상태를 지웁니다.
print 싱크의 결과 줄에서 맨 앞의 w2>는 print-identifier로 붙인 이름이고, +I는 새 행이 추가됐다는 뜻이에요.
그럼 잡을 제출하고 결과를 지켜보면서 10:00:00부터 10:00:09까지 10건을 보내 볼게요. 그다음 벽시계로 60초를 기다리고 워터마크를 봅니다. before.txt에는 00부터 09까지 한 줄에 하나씩 들어 있어요.
submit s1.sql
watch_results &
send clicks1 0 before.txt
date +%T; sleep 60; date +%T
wm
Job ID: 9417a670e7f7d53cef3fc025bc40c794
소스 준비 완료
정답: [00~10) 10건
보냄 #01 10:00:00
보냄 #02 10:00:01
보냄 #03 10:00:02
보냄 #04 10:00:03
보냄 #05 10:00:04
보냄 #06 10:00:05
보냄 #07 10:00:06
보냄 #08 10:00:07
보냄 #09 10:00:08
보냄 #10 10:00:09
10:23:15
10:24:15
LocalWindowAggregate[3].currentInputWatermark: 10:00:07
Source__clicks[1].currentOutputWatermark: 10:00:07
Source__clicks[1].split.clicks1-0.watermark.currentWatermark: 10:00:07
GlobalWindowAggregate[5].currentInputWatermark: 10:00:07
[00~10) 윈도우의 이벤트 10건이 모두 도착했고 벽시계로 1분이 지났지만, 결과는 하나도 나오지 않았어요.
워터마크는 가장 큰 이벤트 시간 09초에서 2초를 뺀 07초에 머물러 있습니다. 파티션별 워터마크(split.clicks1-0), 소스가 내보낸 워터마크, 윈도우 연산자가 받은 워터마크가 모두 같아요.
이제 10:00:12 이벤트 1건(after.txt, 12 한 줄)을 보내 볼게요.
send clicks1 0 after.txt
정답: [10~20) 1건
보냄 #01 10:00:12
w2> +I[2026-10-01T10:00, 2026-10-01T10:00:10, 10]
12초 이벤트가 들어오자 워터마크가 12 - 2 = 10초가 됐고, 곧바로 [00~10) 윈도우의 결과 10건이 나왔어요.
이벤트 시간 윈도우는 벽시계가 아니라 이벤트가 워터마크를 밀어 올려야 닫힙니다. 새 이벤트가 오지 않으면 몇 분이 지나도 마지막 윈도우는 열린 채로 남아요.
실습 2: 지연값에 따라 달라지는 집계
이번에는 순서가 섞인 이벤트를 보내 볼게요. events.txt의 41건은 10초 윈도우 4개에 10건씩, 마지막에 워터마크를 끝까지 밀어 올릴 59초 이벤트 1건이에요. 그중 네 건은 순서를 바꿔 늦게 넣었습니다.
- #07 03초: 06초 다음에 도착해요. 3초 늦었지만 같은 [00~10) 윈도우 안입니다.
- #12 08초: 11초 다음에 도착해요. 3초 늦었고, 이미 [10~20) 윈도우 이벤트가 온 뒤입니다.
- #26 17초: 25초 다음에 도착해요. 8초 늦었습니다.
- #33 29초: 32초 다음에 도착해요. 3초 늦었습니다.
sql/s2.sql은 같은 clicks1 토픽을 읽는 테이블 세 개를 지연값만 0초, 5초, 10초로 다르게 만들고, 결과도 d0, d5, d10으로 나눠 찍어요.
세 집계는 EXECUTE STATEMENT SET으로 묶어서 잡 하나로 제출합니다. 한 번 보낸 이벤트를 세 지연값이 같은 순서로 받으니까 결과를 바로 비교할 수 있어요.
submit이 앞 잡을 취소하고 새 잡을 띄웁니다. 앞에서 띄운 watch_results는 그대로 결과를 보여 줘요.
submit s2.sql
send clicks1 0 events.txt
취소: 9417a670e7f7d53cef3fc025bc40c794
Job ID: 87fd184fcbf31f155f06682ef3319d62
소스 준비 완료
정답: [00~10) 10건, [10~20) 10건, [20~30) 10건, [30~40) 10건, [50~60) 1건
보냄 #01 10:00:00
보냄 #02 10:00:01
보냄 #03 10:00:02
보냄 #04 10:00:04
보냄 #05 10:00:05
보냄 #06 10:00:06
보냄 #07 10:00:03 <- 3초 늦음
보냄 #08 10:00:07
보냄 #09 10:00:09
보냄 #10 10:00:10
d0> +I[2026-10-01T10:00, 2026-10-01T10:00:10, 9]
보냄 #11 10:00:11
보냄 #12 10:00:08 <- 3초 늦음
보냄 #13 10:00:12
보냄 #14 10:00:13
보냄 #15 10:00:14
보냄 #16 10:00:15
d5> +I[2026-10-01T10:00, 2026-10-01T10:00:10, 10]
보냄 #17 10:00:16
보냄 #18 10:00:18
보냄 #19 10:00:19
보냄 #20 10:00:20
d0> +I[2026-10-01T10:00:10, 2026-10-01T10:00:20, 9]
d10> +I[2026-10-01T10:00, 2026-10-01T10:00:10, 10]
보냄 #21 10:00:21
보냄 #22 10:00:22
보냄 #23 10:00:23
보냄 #24 10:00:24
보냄 #25 10:00:25
d5> +I[2026-10-01T10:00:10, 2026-10-01T10:00:20, 9]
보냄 #26 10:00:17 <- 8초 늦음
보냄 #27 10:00:26
보냄 #28 10:00:27
보냄 #29 10:00:28
보냄 #30 10:00:30
d0> +I[2026-10-01T10:00:20, 2026-10-01T10:00:30, 9]
d10> +I[2026-10-01T10:00:10, 2026-10-01T10:00:20, 10]
보냄 #31 10:00:31
보냄 #32 10:00:32
보냄 #33 10:00:29 <- 3초 늦음
보냄 #34 10:00:33
보냄 #35 10:00:34
보냄 #36 10:00:35
d5> +I[2026-10-01T10:00:20, 2026-10-01T10:00:30, 10]
보냄 #37 10:00:36
보냄 #38 10:00:37
보냄 #39 10:00:38
보냄 #40 10:00:39
보냄 #41 10:00:59
d5> +I[2026-10-01T10:00:30, 2026-10-01T10:00:40, 10]
d0> +I[2026-10-01T10:00:30, 2026-10-01T10:00:40, 10]
d10> +I[2026-10-01T10:00:20, 2026-10-01T10:00:30, 10]
d10> +I[2026-10-01T10:00:30, 2026-10-01T10:00:40, 10]
지연값별 윈도우 건수를 정리하면 다음과 같아요.
- 지연값 0초: 9, 9, 9, 10. 08초, 17초, 29초 이벤트가 빠졌습니다.
- 지연값 5초: 10, 9, 10, 10. 8초 늦은 17초 이벤트만 빠졌습니다.
- 지연값 10초: 10, 10, 10, 10. 정답과 같습니다.
같은 실습을 세 번 반복했고, 세 번 모두 윈도우별 건수가 위와 같았어요.
같은 이벤트인데 왜 지연값마다 결과가 다를까요? 지연값 0초에서 #12 08초 이벤트가 도착했을 때 워터마크는 이미 11초였고, [00~10) 윈도우는 그 전에 10초 이벤트로 결과를 냈어요. 지연값 5초에서는 같은 시점의 워터마크가 11 - 5 = 6초라서 [00~10)이 아직 열려 있었습니다.
또 #26 17초 이벤트가 도착했을 때 지연값 5초의 워터마크는 25 - 5 = 20초로 [10~20) 윈도우가 이미 닫혔고, 지연값 10초에서는 15초라서 들어갔어요. 반대로 #07 03초 이벤트는 지연값 0초에서도 집계에 들어갔습니다. 늦게 도착했어도 그 이벤트의 윈도우가 아직 결과를 내기 전이면 빠지지 않아요.
지연값을 키울수록 늦은 이벤트를 더 받아 집계가 정확해지지만, 그만큼 이벤트가 더 들어와야 결과가 나와요. [00~10) 윈도우의 결과는 지연값 0초에서 #10 10초 이벤트, 5초에서 #16 15초 이벤트, 10초에서 #20 20초 이벤트를 보낸 뒤에 나왔습니다. 지연값을 정하는 일은 정확도와 결과 지연 시간 사이에서 고르는 일입니다.
실습 3: 늦은 이벤트를 따로 모으기
지연값을 아무리 크게 잡아도 그보다 늦게 오는 이벤트는 생길 수 있어요. 기본 동작은 집계에서 조용히 빼는 것이라서, 얼마나 빠졌는지 알려면 늦은 이벤트를 따로 모아야 합니다.
Flink SQL에는 DataStream API의 side output 같은 별도 출력이 없어요. 대신 CURRENT_WATERMARK(ts) 함수로 지금 연산자가 받은 워터마크를 읽어 늦은 행을 걸러 낼 수 있습니다. Flink 문서는 이 함수가 워터마크가 아직 없으면 NULL을 돌려준다고 설명하니까 IS NOT NULL 조건을 같이 써 줄게요.
sql/s3.sql은 지연값 0초 테이블 하나에서 세 가지를 같은 잡으로 실행해요.
counts_d0: 앞과 같은 10초 윈도우 집계late_events: 이벤트 시간이 워터마크보다 크지 않은 행. 문서 예시의 조건(ts > CURRENT_WATERMARK(ts))을 뒤집은 것입니다.dropped_events: 그 이벤트가 속한 윈도우의 끝이 이미 워터마크 이하인 행. 윈도우 TVF(앞의TUMBLE(TABLE ...)처럼 테이블을 받아 윈도우 열을 붙인 테이블을 돌려주는 함수)를 집계 없이 쓰면 행마다window_start,window_end열이 붙으니까 이 값으로 비교합니다.
같은 events.txt를 보내 볼게요. 출력에서 잡 제출 메시지와 늦지 않은 이벤트의 보냄 줄은 생략했어요.
submit s3.sql
send clicks1 0 events.txt
정답: [00~10) 10건, [10~20) 10건, [20~30) 10건, [30~40) 10건, [50~60) 1건
보냄 #07 10:00:03 <- 3초 늦음
late> +I[u2, 2026-10-01T10:00:03, 7, 2026-10-01T10:00:06]
d0> +I[2026-10-01T10:00, 2026-10-01T10:00:10, 9]
보냄 #12 10:00:08 <- 3초 늦음
late> +I[u1, 2026-10-01T10:00:08, 12, 2026-10-01T10:00:11]
dropped> +I[u1, 2026-10-01T10:00:08, 12, 2026-10-01T10:00:11]
d0> +I[2026-10-01T10:00:10, 2026-10-01T10:00:20, 9]
보냄 #26 10:00:17 <- 8초 늦음
late> +I[u3, 2026-10-01T10:00:17, 26, 2026-10-01T10:00:25]
dropped> +I[u3, 2026-10-01T10:00:17, 26, 2026-10-01T10:00:25]
d0> +I[2026-10-01T10:00:20, 2026-10-01T10:00:30, 9]
보냄 #33 10:00:29 <- 3초 늦음
late> +I[u1, 2026-10-01T10:00:29, 33, 2026-10-01T10:00:32]
dropped> +I[u1, 2026-10-01T10:00:29, 33, 2026-10-01T10:00:32]
d0> +I[2026-10-01T10:00:30, 2026-10-01T10:00:40, 10]
출력을 보면 late에는 4건, dropped에는 3건이 찍혔어요. dropped의 3건(seq 12, 26, 33)은 앞에서 본 지연값 0초 집계에서 빠진 08초, 17초, 29초 이벤트와 정확히 같습니다. 마지막 열은 그 행을 처리할 때의 워터마크예요.
late에만 있는 #07 03초 이벤트는 워터마크(06초)보다 늦게 왔지만, [00~10) 윈도우가 아직 열려 있어 집계에 들어갔어요.
정리하면 ts <= CURRENT_WATERMARK(ts) 조건은 "순서가 뒤집힌 이벤트"를 모두 고르고, 윈도우 집계가 실제로 버리는 이벤트는 "결과를 이미 낸 윈도우의 이벤트"예요. 그래서 빠진 데이터를 다시 반영하려는 목적이라면 윈도우 끝과 비교하는 쪽을 써야 중복으로 더하지 않습니다.
정리
범위: Flink 2.2.1(flink:2.2.1-java17), Kafka 4.3.1, Flink SQL, 이벤트 41건 중 늦게 넣은 4건, 지연값 비교는 3회 반복.
- [00~10) 이벤트 10건이 모두 도착하고 벽시계로 1분이 지나도 결과가 없다가, 12초 이벤트가 워터마크를 10초로 올리자 결과가 나왔습니다. 이벤트가 끊기는 시간대가 있으면 마지막 윈도우가 열린 채 남는다고 보고 결과 요구 시간을 정하세요.
- 같은 이벤트에서 지연값 0초는 3건, 5초는 1건을 집계에서 뺐고 10초는 정답과 같았으며, 세 번 반복해도 윈도우별 건수가 같았습니다. 지연값은 받아들일 늦은 이벤트의 범위를 보고 고르세요.
ts <= CURRENT_WATERMARK(ts)는 4건을 골랐고, 그중 3초 늦은 #07은 윈도우가 열려 있어 집계에 들어간 이벤트였습니다.window_end <= CURRENT_WATERMARK(ts)가 고른 3건은 집계에서 빠진 3건과 정확히 같았으니, 빠진 데이터를 다시 반영하려면 윈도우 끝과 비교하는 조건으로 모으세요.