동적 테이블과 변경 로그 — 스트리밍 결과는 고쳐 쓰이는 표다

한 줄 요약

Flink SQL 에서 스트림은 계속 행이 추가되는 표(동적 테이블) 이고, 그 위의 질의 결과도 표다. 결과 표가 바뀔 때마다 엔진은 변경 로그(+I 추가 · -U 옛 값 철회 · +U 새 값 · -D 삭제)를 내보낸다. 어느 연산자가 어떤 종류의 변경을 내는지가 곧 어느 싱크에 쓸 수 있는지를 정한다.

왜 이게 필요했나

배치 SQL 은 입력이 다 모인 뒤 한 번 돌고 끝난다. 결과는 한 번 쓰면 그만이다. 스트림에는 "다 모인 뒤" 가 없다. 사용자별 클릭 수를 세는 질의를 스트림에 걸면, 첫 클릭이 왔을 때 u04 → 1 이라고 답했다가 두 번째 클릭이 오면 그 답을 고쳐야 한다. 이미 내보낸 답을 어떻게 고칠 것인가 — 이것이 스트리밍 SQL 의 핵심 문제다.

Flink 가 택한 답은 데이터베이스의 구체화 뷰(materialized view)를 빌려 오는 것이다. 공식 문서(Dynamic Tables)는 이렇게 정리한다. 데이터베이스의 표는 INSERT·UPDATE·DELETE 의 흐름(changelog stream)으로 만들어지고, 구체화 뷰는 그 흐름을 받아 결과를 계속 고쳐 쓴다. 스트림을 표로, 연속 질의를 뷰 유지로 보면 SQL 의 의미를 그대로 스트림에 옮길 수 있다. 그리고 문서의 약속이 하나 있다 — 연속 질의의 결과는 어느 순간이든 그 시점의 입력 스냅숏에 같은 질의를 배치로 돌린 결과와 의미가 같다. 이 모듈의 실습은 이 약속을 직접 확인하는 데서 출발한다.

어떻게 동작하나

스트림 → 표. 원천의 각 레코드는 결과 표에 대한 INSERT 로 해석된다. 파일이나 로그에서 읽은 클릭은 추가만 되는(append-only) 표다.

표 → 표. 질의는 입력 표가 바뀔 때마다 결과 표를 고친다. 필터(WHERE dwell > 60)나 투영은 입력 한 행이 결과 한 행이 되고 끝이라, 결과도 추가만 된다. 반면 GROUP BY user_id 의 COUNT(*) 는 같은 키의 행이 올 때마다 이미 낸 결과 행을 바꿔야 한다. 문서는 앞의 것을 append 질의, 뒤의 것을 update 질의라 부르고, update 질의는 이미 낸 결과를 고치려고 더 많은 상태를 쥐어야 한다고 적는다.

표 → 스트림. 결과 표의 변화를 밖으로 내보낼 때는 인코딩을 골라야 한다. 문서가 드는 세 가지다.

| 인코딩 | 무엇을 보내나 | 받는 쪽에 필요한 것 |
| --- | --- | --- |
| append-only | 추가된 행만 | 없음 — 결과가 추가만 될 때만 가능 |
| retract | INSERT 는 추가, DELETE 는 철회, UPDATE 는 철회(옛 행) + 추가(새 행) 두 메시지 | 없음 — 옛 행 값을 받아 그대로 지운다 |
| upsert | INSERT·UPDATE 는 upsert 한 메시지, DELETE 는 삭제 | 고유 키 — 키로 찾아 덮어쓴다 |

sql-client 의 표 모드 출력에 보이는 op 열이 바로 이 변경 종류다. +I 추가, -U 갱신 전 값의 철회, +U 갱신 후 값, -D 삭제. 사용자별 COUNT 라면 새 사용자는 +I 한 줄, 기존 사용자는 -U·+U 두 줄을 낸다. 입력이 추가만 되고 병렬도가 1 이면 이 로그의 순서와 개수는 입력 순서로 완전히 정해진다 — 실습 채점기가 로그를 파이썬으로 한 줄씩 재현해 맞춰 보는 근거다.

집계 위에 집계를 쌓으면 삭제도 생긴다. "클릭 수가 n 번인 사용자가 몇 명인가" 를 세면, 한 사용자가 1번 칸에서 2번 칸으로 옮길 때 안쪽 집계의 -U 가 바깥 1번 칸을 하나 줄인다. 그 칸이 0 이 되면 결과 행 자체가 사라져야 하므로 -D 가 나온다.

연산자마다 어떤 변경을 내는지는 EXPLAIN CHANGELOG_MODE 가 계획에 찍어 준다.

GroupAggregate(groupBy=[user_id], ..., changelogMode=[I,UB,UA])   ← 갱신 전·후를 다 낸다GroupAggregate(groupBy=[user_id], ..., changelogMode=[I,UA])      ← 같은 집계, 업서트 싱크 앞Calc(select=[user_id, url], where=[>(dwell, 60)], changelogMode=[I])

같은 집계인데 모드가 둘인 이유는 싱크가 무엇을 받을 수 있느냐를 옵티마이저가 거꾸로 거슬러 올라가며 정하기 때문이다. 키 없는 print 싱크는 철회가 있어야 옛 행을 지울 수 있으므로 UB 를 요구하고, PRIMARY KEY 가 선언된 싱크는 키로 덮어쓰면 되므로 UB 를 뺀다. 메시지 수가 거의 절반으로 준다. 반대로 filesystem 싱크처럼 한 번 쓴 줄을 고칠 수 없는 싱크에 update 질의를 넣으면 잡이 뜨기도 전에 "doesn't support consuming update changes" 로 거절된다. 그래도 로그를 파일로 남기고 싶다면 TO_CHANGELOG 로 변경 종류를 열로 꺼내 모든 행을 추가로 바꾸면 된다(문서의 Changelog Conversion).

현장에서 만나는 모습

가장 흔한 사고는 대시보드 숫자가 두 배로 부풀어 있는 것이다. 리트랙트 스트림을 받는 쪽이 -U 를 무시하고 +U 만 더했거나, 반대로 추가만 하는 저장소에 로그를 그대로 붙였기 때문이다. op 를 보지 않고 행만 세면 갱신이 곧 중복이 된다. 결과 표의 정의(키가 무엇인지)와 싱크의 쓰기 방식(덮어쓰기인지 추가인지)을 짝지어 확인하는 것이 먼저다.

두 번째는 "스트리밍 결과가 배치와 다르다" 는 신고다. 최종 상태를 비교하면 대개 같다. 다른 것은 중간 로그를 누가 어떻게 받았느냐다. 다만 결과가 같다는 보장은 입력이 같을 때의 이야기다. 시간에 따라 버리는 행이 있는 질의(다음 모듈의 워터마크)나 상태를 시간으로 지우는 설정이 끼면 배치와 달라질 수 있다.

세 번째는 업서트 싱크의 키가 결과 표의 키와 어긋난 경우다. 집계 키는 user_id 인데 싱크의 PRIMARY KEY 를 다른 열로 잡으면, 싱크가 받는 변경이 싱크 키 기준으로 순서가 뒤섞일 수 있다. 문서(Configuration 의 table.exec.sink.upsert-materialize)는 이런 뒤섞임이 생기면 옵티마이저가 싱크 앞에 upsert materialize 연산자를 끼운다고 적는다 — 키마다 받은 행을 상태로 쥐고 있다가 올바른 upsert 순서로 다시 내는 연산자다. 상태가 하나 더 느는 셈이라, 싱크 키는 처음부터 결과 표의 고유 키와 같게 잡는 것이 원칙이다.

다음 실습에서 할 것

클릭 120건을 사용자별로 세는 질의를 배치와 스트리밍으로 각각 돌려 최종 상태가 같다는 것과 스트리밍 로그 229줄을 확인한다. 집계 위의 집계에서 -D 가 나오는 것을 보고, EXPLAIN CHANGELOG_MODE 로 집계와 필터의 변경 종류를 비교한다. 집계를 파일 싱크에 넣어 거절당한 뒤 TO_CHANGELOG 로 바꿔 파일에 쓰고, 키 없는 싱크와 키 있는 싱크 앞에서 계획이 어떻게 달라지는지 본다. 마지막으로 같은 로그를 업서트로 보냈다면 메시지가 몇 건이었을지 보고서에 적는다.