Apache Flink — 스트림을 엔진으로 돌린다 · 중복 제거와 Top-N · 讲解

중복 제거와 Top-N — 같은 ROW_NUMBER, 다른 변경 로그

한 줄 요약

Flink SQL 의 중복 제거와 Top-N 은 둘 다 ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) 에 순번 조건을 붙인 같은 패턴이다. 그런데 첫 행을 남기면 결과가 append-only이고, 마지막 행을 남기거나 순위를 매기면 이미 낸 행을 철회하고 고치는 갱신 로그가 된다. 어느 쪽인지가 하류의 싱크와 집계를 정한다.

概念图: 같은 패턴 · 첫 행을 남기면 결과가 append-only · 마지막 행을 남기거나 순위를 매기면 이미 낸 행을 철회하고 고치는 갱신 로그 · 같은 사건이 두 번 온 것

왜 이게 필요했나

공식 문서가 드는 예가 현실 그대로다. 상류 ETL 이 끝에서 끝까지 정확히 한 번을 보장하지 못하면, 장애 복구 때 같은 레코드가 싱크에 두 번 들어간다. 그 상태로 SUM·COUNT 를 하면 숫자가 부풀어 오른다. 그래서 분석 전에 중복을 걷어 내야 한다.

그런데 "중복" 에는 두 가지 뜻이 섞여 있다. 하나는 같은 사건이 두 번 온 것(재전송) — 처음 것 하나만 남기면 된다. 다른 하나는 같은 대상의 상태가 여러 번 바뀐 것(주문이 생성 → 결제 → 배송) — 지금 상태, 곧 마지막 것을 남겨야 한다. 두 요구는 SQL 로는 ASC 와 DESC 한 단어 차이지만, 엔진 안에서는 전혀 다른 일을 한다. 첫 행은 한 번 내면 끝이지만, 마지막 행은 "지금까지의 마지막" 이 계속 바뀌기 때문이다.

어떻게 동작하나

왼쪽은 주문 o0018 의 이벤트 세 줄(created, paid, 같은 paid 재전송)이 들어올 때 두 연산자가 내는 로그다. 첫 행 유지는 created 를 +I 로 한 번 내고 끝이다. 마지막 행 유지는 created 를 +I 로 낸 뒤 paid 가 오면 -U created 와 +U paid 한 쌍을, 재전송 paid 가 와도 -U paid 와 +U paid 한 쌍을 더 낸다. 오른쪽은 문서의 Top-N 예로, 9위 상품이 1위로 올라가면 순번을 내보내는 쿼리는 1위부터 9위까지 아홉 행을 모두 다시 보내지만 순번을 빼면 바뀐 상품 하나만 보낸다

패턴을 그대로 지켜야 한다. 문서는 중복 제거를 ROW_NUMBER() · PARTITION BY 키 · ORDER BY 시간 속성 · 바깥의 WHERE rownum = 1 로 정의하고, 이 모양을 정확히 따라야 옵티마이저가 알아본다고 적는다. ORDER BY 는 반드시 시간 속성(처리 시간 또는 이벤트 시간)이어야 하고, ASC 는 첫 행, DESC 는 마지막 행을 남긴다. 이론적으로 중복 제거는 N 이 1 이고 시간으로 정렬한 Top-N 의 특수한 경우다.

계획에서 보면 이렇게 갈린다. EXPLAIN CHANGELOG_MODE 의 물리 계획은 이 연산자를 Rank(strategy=[AppendFastStrategy], rankRange=[rankStart=1, rankEnd=1], ...) 로 적고 줄 끝에 내보내는 변경 종류를 붙인다. 실행 계획에서는 같은 자리가 Deduplicate 노드로 바뀐다.

물리 계획  Rank(... orderBy=[ROWTIME ts ASC] ...)   changelogMode=[I]실행 계획  Deduplicate(keep=[FirstRow], key=[order_id], order=[ROWTIME], outputInsertOnly=[true])물리 계획  Rank(... orderBy=[ROWTIME ts DESC] ...)  changelogMode=[I,UA,D]실행 계획  Deduplicate(keep=[LastRow], key=[order_id], order=[ROWTIME], outputInsertOnly=[false])

첫 행 유지는 키마다 "이미 봤다" 는 표시만 상태에 두고, 두 번째부터는 버린다. 결과는 insert-only 라 파일 같은 append-only 싱크에도 그대로 들어간다. 마지막 행 유지는 키마다 지금까지의 마지막 행을 상태에 두고, 새 행이 오면 앞서 낸 행을 철회하고 새 행을 낸다. 실측에서 이벤트 622행·주문 240건을 넣었더니 +I 240개와 -U/+U 382쌍이 나왔다. 382 = 622 − 240, 곧 같은 키의 두 번째 이후 행마다 한 쌍이다. 완전히 같은 행이 재전송돼도 쌍을 냈다. 처리 시간(PROCTIME())으로 정렬해 보면 같은 입력에서 재전송 41건 중 18건의 쌍이 나오지 않았다 — 도착 시각에 기대는 정렬이라 이런 개수는 실행 환경에 따라 달라질 수 있어, 실습은 이벤트 시간으로 판정한다.

Top-N 은 순번 조건이 <= N 인 같은 패턴이다. 문서는 Top-N 이 결과 갱신형이라, 상위 N 이 바뀌면 바뀐 행을 철회·갱신으로 보낸다고 적는다. 입력 자체가 갱신되는 경우(판매량 SUM 위의 순위)라면 계획에 Rank(strategy=[RetractStrategy], ...) 가 나온다. 여기서 중요한 선택이 순번을 내보내느냐다. 순번 칸은 결과의 유일 키 일부가 되므로, 문서의 예처럼 9위가 1위로 오르면 1–9위 행이 모두 다시 나간다. 바깥 SELECT 에서 순번을 빼면 바뀐 상품 하나만 보내면 된다. 실측에서 같은 판매 파일의 범주별 Top-3 는 순번을 내보냈을 때 로그 3,419행, 뺐을 때 1,523행이었고 최종 결과는 같았다.

마지막 행 유지 위에 GROUP BY status 를 얹으면 두 연산자가 맞물린다. 주문 하나가 created 에서 paid 로 바뀌면 중복 제거가 -U created·+U paid 를 내고, 집계는 그것을 받아 created 칸에서 하나를 빼고 paid 칸에 하나를 더한다. 중복 제거 없이 이벤트를 그대로 세면 한 주문이 여러 칸에 동시에 잡힌다.

앞 모듈의 temporal join 에서 버전 뷰를 만든 것도 바로 이 마지막 행 유지다. 창 단위 Top-N 은 창 모듈에서 다뤘다 — 거기서는 창이 닫힐 때 한 번만 내므로 철회가 없다.

현장에서 만나는 모습

가장 흔한 실수는 "중복 제거 결과를 파일이나 append-only 토픽에 쓴다" 다. 첫 행 유지면 문제없지만, 마지막 행 유지로 바꾸는 순간 결과가 갱신 로그가 되어, 갱신을 받지 못하는 싱크에는 넣을 수 없다(동적 테이블 모듈에서 본 거절이다). 둘 중 무엇이 필요한지 요구부터 가르는 것이 먼저다. 재전송을 걷어 내는 거라면 첫 행, 지금 상태가 필요하면 마지막 행이다.

두 번째는 Top-N 의 로그 폭주다. 실시간 순위표를 키-값 저장소에 쓰는데 쓰기량이 예상의 몇 배라면 순번을 내보내고 있는지부터 본다. 화면이 순번을 스스로 매길 수 있다면 순번을 빼는 것만으로 쓰기가 크게 준다.

세 번째는 정렬 기준이다. 처리 시간으로 중복 제거하면 결과가 도착 순서에 달려 다시 돌릴 때마다 다를 수 있다. 재처리해도 같은 답이 나와야 하는 파이프라인이라면 이벤트 시간을 쓴다.

다음 실습에서 할 것

주문 상태 이벤트와 판매 기록을 정의하고 행 수와 주문 수를 센다. 첫 행 유지와 마지막 행 유지를 돌려 로그 개수가 어떻게 다른지 확인하고, EXPLAIN 으로 두 계획의 변경 로그 모드를 비교한다. 마지막 행 유지 위에 상태별 현황을 세고, 범주별 Top-3 를 순번과 함께·순번 없이 돌려 로그 양을 비교한 뒤 보고서로 정리한다.