마이크로배치 집계

날짜 전체를 매번 다시 읽는 방식은 단순하지만 원본이 커지면 비용이 큽니다. 그렇다고 이벤트의 조회수를 일별 합계에 바로 더하면 같은 이벤트 재전송, 게시글 시간 정정, 날짜 이동에서 빼기와 더하기 순서를 관리해야 합니다. 체크포인트 숫자만 저장해도 대상 반영과 따로 커밋하면 유실 또는 중복이 생깁니다.

마이크로배치의 안전한 단위는 이벤트 증가분이 아니라 “각 게시글이 현재 통계에 기여하는 절대값”입니다. post_id별 기여값을 최신 source_version으로 덮고, 바뀐 회원/날짜 키만 기여값 전체에서 다시 합산합니다. 체크포인트, 기여값, 일별 대상을 하나의 트랜잭션으로 전진시킵니다.

post_change_event는 추가 전용 큐입니다. 완료 게시글은 건수=1과 조회수를, 취소·삭제 게시글은 건수=0과 조회수=0을 싣습니다. 같은 게시글의 정정은 source_version을 올립니다.

소비자 이름별 체크포인트는 마지막 완료 event_id를 저장합니다. 배치는 다음 N개 이벤트의 상한을 정하고 그 범위에서 게시글별 가장 최신 이벤트만 사용합니다. 중간 버전을 생략해도 절대 기여가 최종 상태를 표현하므로 결과가 같습니다.

집계와 체크포인트의 분리 저장

일별 UPDATE가 커밋된 뒤 체크포인트 UPDATE 전에 장애가 나면 같은 이벤트가 다시 더해집니다. 순서를 바꾸면 체크포인트만 전진하고 합계가 빠질 수 있습니다. 두 테이블을 한 트랜잭션에 넣어도 정정 이벤트가 +90이라면 기존 +60을 제거하지 못합니다.

재소비와 정정에서 drift가 생기는 delta consumer

DROP TABLE IF EXISTS final_stat_delta_bad;
CREATE TABLE final_stat_delta_bad(
 author_id BIGINT UNSIGNED,stat_date DATE,total_views BIGINT UNSIGNED,
 PRIMARY KEY(author_id,stat_date)
) ENGINE=InnoDB;
INSERT INTO final_stat_delta_bad VALUES(1,'2026-07-15',60);
-- event 101 처리 뒤 응답을 잃어 동일 payload를 재소비한다.
UPDATE final_stat_delta_bad SET total_views=total_views+60
WHERE author_id=1 AND stat_date='2026-07-15';
-- 실제 게시글이 조회수 90으로 정정됐지만 consumer는 새 90을 다시 더한다.
UPDATE final_stat_delta_bad SET total_views=total_views+90
WHERE author_id=1 AND stat_date='2026-07-15';
SELECT * FROM final_stat_delta_bad;

한 게시글의 현재 기여는 조회수 90인데 대상에는 조회수 210이 남습니다.

오류 실행 결과

source posts=1
current source views=90
delta target views=210
duplicate replay excess=60
correction replacement missing=60

변화량이 정확하려면 이전 기여를 알고 신규-이전 차이를 한 번만 적용해야 합니다. 이벤트 소비자가 그 상태를 가지지 않으면 단순 덧셈은 수정과 삭제를 처리할 수 없습니다. 전달 정확히 한 번을 브로커에 기대해도 DB 커밋 응답 경계는 남습니다.

게시글 기여값 테이블은 이 문제를 업무 집계 단위로 바꿉니다. source_version이 더 큰 이벤트만 게시글 행을 교체합니다. 일별 합계는 기여값을 SUM한 절대 결과이므로 이벤트가 재등장해도 값이 증가하지 않습니다.

이전 키와 신규 키를 모두 변경 대상으로 잡기

게시글의 published_at 또는 회원이 바뀌면 기여값이 다른 날짜로 이동합니다. UPDATE 전에 기존 회원/날짜를 affected_keys에 넣고, 이벤트의 새 회원/날짜도 넣어야 이전 날짜에서 값이 빠집니다.

v_from은 저장 체크포인트, v_to는 이번에 선택한 마지막 이벤트 ID입니다. 트랜잭션이 성공하면 기여값, 영향받은 일별 행, 체크포인트가 한 커밋에서 함께 확정됩니다. 체크포인트는 v_to까지 전진하고 기여값은 창 안에서 선택된 최고 버전의 event_id를 기록합니다. 롤백되면 체크포인트는 v_from에 머물고 기여값과 일별 행은 직전 커밋 상태로 남습니다. 연결 응답을 잃어 재호출해도 DB 상태가 어느 경계인지 체크포인트로 판단됩니다.

배치 안에서 같은 게시글 이벤트가 여러 건이면 가장 큰 source_version을 선택합니다. 버전이 같다면 가장 큰 event_id를 선택하는 방어적 동률 규칙을 둡니다. 현재 원장은 유일 제약으로 같은 게시글·버전의 중복 저장을 먼저 막지만, 재생성 입력을 합칠 때도 이 정렬 규칙을 그대로 사용할 수 있습니다. 순서를 벗어나 도착한 낮은 버전은 체크포인트에는 포함되지만 기여값을 되돌리지 않습니다.

대상은 기여값의 캐시이지 이벤트 원장이 아니다

이벤트 보존 기간이 끝나도 기여값은 현재 원본 상태를 나타냅니다. 장애 복구 시 posts에서 기여값 전체를 재구축할 수 있어야 합니다. 일별 행은 기여값에서 다시 만들어집니다.

MySQL 8.4 메모: 사용 중단 예정 VALUES() UPSERT를 피하려고 임시 next_contribution과 명시적 UPDATE/INSERT 두 단계를 사용합니다. 두 단계는 동일 트랜잭션과 PK 잠금 안에 있습니다.

최신 기여값 기반 집계

DDL은 이벤트 업무 버전 유일성, 기여값 PK, 소비자 체크포인트를 정의합니다. 프로시저는 임시 테이블 두 개를 사용합니다. next_contribution은 최종 적용할 이벤트 내용을, affected_stat_key는 이전·신규 날짜를 담고 v_to는 처리 상한을 나타냅니다. 공개 게시글도 조회수 0일 수 있으므로 published_count=1 분기는 total_views 0부터 허용합니다.

checkpoint microbatch의 완전한 schema와 처리 procedure

DROP PROCEDURE IF EXISTS process_post_stats_microbatch;
DROP TABLE IF EXISTS daily_post_stats_incremental;
DROP TABLE IF EXISTS post_stat_contribution;
DROP TABLE IF EXISTS final_stat_checkpoint;
DROP TABLE IF EXISTS post_change_event;

CREATE TABLE post_change_event(
 event_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
 post_id BIGINT UNSIGNED NOT NULL,source_version INT UNSIGNED NOT NULL,
 author_id BIGINT UNSIGNED NOT NULL,stat_date DATE NOT NULL,
 published_count TINYINT UNSIGNED NOT NULL,total_views BIGINT UNSIGNED NOT NULL,
 produced_at DATETIME(6) NOT NULL,
 CONSTRAINT uq_stat_event_version UNIQUE(post_id,source_version),
 CONSTRAINT chk_stat_event_value CHECK(
  (published_count=0 AND total_views=0)
  OR (published_count=1 AND total_views<=1000000000)),
 INDEX ix_stat_event_stream(event_id)
) ENGINE=InnoDB;
CREATE TABLE post_stat_contribution(
 post_id BIGINT UNSIGNED PRIMARY KEY,source_version INT UNSIGNED NOT NULL,
 author_id BIGINT UNSIGNED NOT NULL,stat_date DATE NOT NULL,
 published_count TINYINT UNSIGNED NOT NULL,total_views BIGINT UNSIGNED NOT NULL,
 last_event_id BIGINT UNSIGNED NOT NULL,
 INDEX ix_contribution_day(author_id,stat_date)
) ENGINE=InnoDB;
CREATE TABLE daily_post_stats_incremental(
 author_id BIGINT UNSIGNED NOT NULL,stat_date DATE NOT NULL,
 published_count INT UNSIGNED NOT NULL,total_views BIGINT UNSIGNED NOT NULL,
 rebuilt_from_event BIGINT UNSIGNED NOT NULL,
 PRIMARY KEY(author_id,stat_date)
) ENGINE=InnoDB;
CREATE TABLE final_stat_checkpoint(
 consumer_name VARCHAR(80) PRIMARY KEY,last_event_id BIGINT UNSIGNED NOT NULL,
 updated_at DATETIME(6) NOT NULL
) ENGINE=InnoDB;
INSERT INTO final_stat_checkpoint VALUES('daily-stat-v1',0,'2026-07-16 00:00:00');

DELIMITER //
CREATE PROCEDURE process_post_stats_microbatch(IN p_consumer VARCHAR(80),IN p_limit INT UNSIGNED)
proc: BEGIN
 DECLARE v_from BIGINT UNSIGNED; DECLARE v_to BIGINT UNSIGNED;
 DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN ROLLBACK; DROP TEMPORARY TABLE IF EXISTS next_contribution; DROP TEMPORARY TABLE IF EXISTS affected_stat_key; RESIGNAL; END;
 START TRANSACTION;
 SELECT last_event_id INTO v_from FROM final_stat_checkpoint WHERE consumer_name=p_consumer FOR UPDATE;
 SELECT MAX(event_id) INTO v_to FROM(
  SELECT event_id FROM post_change_event WHERE event_id>v_from ORDER BY event_id LIMIT p_limit
 ) batch_ids;
 IF v_to IS NULL THEN COMMIT; SELECT v_from AS checkpoint,'EMPTY' outcome; LEAVE proc; END IF;

 CREATE TEMPORARY TABLE next_contribution(
  post_id BIGINT UNSIGNED PRIMARY KEY,source_version INT UNSIGNED,author_id BIGINT UNSIGNED,
  stat_date DATE,published_count TINYINT UNSIGNED,total_views BIGINT UNSIGNED,event_id BIGINT UNSIGNED
 ) ENGINE=InnoDB;
 INSERT INTO next_contribution
  (post_id,source_version,author_id,stat_date,published_count,total_views,event_id)
 SELECT post_id,source_version,author_id,stat_date,published_count,total_views,event_id
 FROM(
  SELECT e.*,
   ROW_NUMBER() OVER(
    PARTITION BY e.post_id
    ORDER BY e.source_version DESC,e.event_id DESC
   ) AS version_rank
  FROM post_change_event e
  WHERE e.event_id>v_from AND e.event_id<=v_to
 ) ranked
 WHERE version_rank=1;

 CREATE TEMPORARY TABLE affected_stat_key(
  author_id BIGINT UNSIGNED,stat_date DATE,PRIMARY KEY(author_id,stat_date)
 ) ENGINE=InnoDB;
 INSERT IGNORE INTO affected_stat_key
 SELECT c.author_id,c.stat_date FROM post_stat_contribution c
 JOIN next_contribution n ON n.post_id=c.post_id WHERE n.source_version>c.source_version;
 INSERT IGNORE INTO affected_stat_key
 SELECT n.author_id,n.stat_date FROM next_contribution n
 LEFT JOIN post_stat_contribution c ON c.post_id=n.post_id
 WHERE c.post_id IS NULL OR n.source_version>c.source_version;

 UPDATE post_stat_contribution c JOIN next_contribution n ON n.post_id=c.post_id
 SET c.source_version=n.source_version,c.author_id=n.author_id,c.stat_date=n.stat_date,
     c.published_count=n.published_count,c.total_views=n.total_views,c.last_event_id=n.event_id
 WHERE n.source_version>c.source_version;
 INSERT INTO post_stat_contribution
  (post_id,source_version,author_id,stat_date,published_count,total_views,last_event_id)
 SELECT n.post_id,n.source_version,n.author_id,n.stat_date,n.published_count,n.total_views,n.event_id
 FROM next_contribution n LEFT JOIN post_stat_contribution c ON c.post_id=n.post_id
 WHERE c.post_id IS NULL;

 DELETE d FROM daily_post_stats_incremental d JOIN affected_stat_key a
  ON a.author_id=d.author_id AND a.stat_date=d.stat_date;
 INSERT INTO daily_post_stats_incremental
 SELECT c.author_id,c.stat_date,SUM(c.published_count),SUM(c.total_views),v_to
 FROM post_stat_contribution c JOIN affected_stat_key a
  ON a.author_id=c.author_id AND a.stat_date=c.stat_date
 GROUP BY c.author_id,c.stat_date HAVING SUM(c.published_count)>0;
 UPDATE final_stat_checkpoint SET last_event_id=v_to,updated_at=CURRENT_TIMESTAMP(6)
 WHERE consumer_name=p_consumer;
 COMMIT;
 DROP TEMPORARY TABLE next_contribution; DROP TEMPORARY TABLE affected_stat_key;
 SELECT v_to AS checkpoint,'APPLIED' outcome;
END//
DELIMITER ;

SET @post_id = (
 SELECT id FROM posts WHERE status='PUBLISHED' ORDER BY id LIMIT 1
);
SET @author_id = (SELECT author_id FROM posts WHERE id=@post_id);
INSERT INTO post_change_event
 (post_id,source_version,author_id,stat_date,published_count,total_views,produced_at)
VALUES(@post_id,1,@author_id,'2026-07-15',1,60,'2026-07-16 01:00:00');
CALL process_post_stats_microbatch('daily-stat-v1',100);

첫 배치에서는 이벤트가 한 건이므로 체크포인트, 기여값, 일별 행에 이벤트 1의 상태가 함께 반영됩니다.

개선 실행 결과

checkpoint=1
post contribution=60
daily total=60
immediate retry outcome=EMPTY
daily total after retry=60

낮은 source_version 이벤트는 체크포인트 범위에는 포함되지만 기여값을 되돌리지 않습니다. 생산자 버그를 숨기지 않도록 별도 쿼리로 event_id 순서와 source_version 역전을 경고합니다.

DELETE 이벤트도 기여값 행을 없애지 않고 0 값으로 남깁니다. 그래야 오래된 이벤트가 다시 와도 source_version 비교가 가능합니다. 필요하면 삭제 표식 보존 기간 이후 전체 재생성과 함께 정리합니다.

배치 크기는 정확성이 아니라 운영 비용을 조절한다

p_limit을 바꿔도 최종 기여값과 일별 값은 같아야 합니다. 큰 배치는 처리량을 높이지만 체크포인트 행 잠금과 언두 로그가 길어집니다. 처리 지연, 트랜잭션 시간, 영향받은 키 수를 함께 관찰합니다.

정정·이동·응답 손실 검증

버전 3은 조회수를 90으로 바꾸면서 다음 날짜로 옮기고, 그 뒤 더 낮은 버전 2가 늦게 도착합니다. 한 배치가 두 이벤트를 함께 읽어도 최고 source_version인 버전 3을 적용해야 합니다. 같은 프로시저를 다시 호출해 EMPTY가 되는지도 봅니다.

microbatch 상태 전이와 source-target 대사

INSERT INTO post_change_event
 (post_id,source_version,author_id,stat_date,published_count,total_views,produced_at)
VALUES
 (@post_id,3,@author_id,'2026-07-16',1,90,'2026-07-16 03:00:00'),
 (@post_id,2,@author_id,'2026-07-15',1,90,'2026-07-16 02:00:00');
CALL process_post_stats_microbatch('daily-stat-v1',100);
CALL process_post_stats_microbatch('daily-stat-v1',100);

WITH expected AS(
 SELECT author_id,stat_date,SUM(published_count) published_count,SUM(total_views) total_views
 FROM post_stat_contribution GROUP BY author_id,stat_date HAVING SUM(published_count)>0
)
SELECT e.author_id,e.stat_date FROM expected e LEFT JOIN daily_post_stats_incremental d
 ON d.author_id=e.author_id AND d.stat_date=e.stat_date
WHERE d.author_id IS NULL OR d.published_count<>e.published_count OR d.total_views<>e.total_views
UNION ALL
SELECT d.author_id,d.stat_date FROM daily_post_stats_incremental d LEFT JOIN expected e
 ON e.author_id=d.author_id AND e.stat_date=d.stat_date WHERE e.author_id IS NULL;

SELECT e.post_id,MAX(e.source_version) expected_version,c.source_version actual_version
FROM post_change_event e JOIN post_stat_contribution c ON c.post_id=e.post_id
GROUP BY e.post_id,c.source_version
HAVING c.source_version<>MAX(e.source_version);

SELECT earlier.event_id AS higher_version_event,later.event_id AS late_lower_version_event
FROM post_change_event earlier JOIN post_change_event later
 ON later.post_id=earlier.post_id AND later.event_id>earlier.event_id
WHERE later.source_version<earlier.source_version;

최종적으로 7월 15일 행은 사라지고 7월 16일에 조회수 90회가 있습니다. 기여값 일치 검사와 최고 버전 불일치 검사는 0행입니다. 마지막 도착 순서 검사는 버전 역전 한 건을 반환해도 최종 기여값이 버전 3임을 보여 줍니다. 마지막 빈 호출은 체크포인트를 바꾸지 않습니다.

프로시저 커밋 직후 클라이언트 연결을 끊어도 재호출은 저장된 체크포인트 다음부터 시작합니다. 트랜잭션 중간에 강제 오류를 넣으면 체크포인트와 대상 모두 이전 값에 남습니다.

여러 소비자의 독립성

주간 통계와 알림이 같은 이벤트를 읽어도 consumer_name별 체크포인트를 가집니다. 한 소비자의 지연이 다른 대상을 막지 않습니다. 이벤트 보존 기간은 모든 필수 소비자의 최소 체크포인트보다 앞선 행을 지우지 않도록 계산합니다.

처리 불가 이벤트 처리

검사를 통과했지만 업무상 해석할 수 없는 이벤트가 있으면 배치가 계속 롤백될 수 있습니다. 오류 이벤트를 오류 보관 테이블로 옮기고 체크포인트를 넘길지, 운영자가 원본을 고칠 때까지 멈출지 중요도별 정책을 정합니다. 조용히 건너뛰기하면 통계 손실이 생깁니다.

전체 재생성 안전망

기여값은 캐시이므로 posts에서 전체 재생하는 유지보수 프로시저가 필요합니다. 재생 결과 체크섬이 현재 기여값과 다르면 이벤트 생산자 누락 또는 소비자 버그를 조사합니다.

마이크로배치 도입 지표

판단 축	확인할 질문
비용	절대값 날짜 재빌드가 SLA 안에 끝나지 않는가?
변경	수정·삭제·날짜 이동 이벤트를 생산자가 버전으로 발행하는가?
원자성	체크포인트와 두 대상 테이블을 같은 DB 트랜잭션에 둘 수 있는가?
지연	이벤트 지연과 가장 오래된 미처리 경과 시간을 측정하는가?
복구	원본에서 기여값을 전부 재생하는 경로가 있는가?

전체 재빌드가 충분히 싸면 ch16-3이 더 읽기 쉽습니다. 원본 규모 때문에 증분이 필요할 때도 변화량을 바로 더하지 말고 게시글 기여값을 둡니다. 설계 선택에는 이벤트 보존 기간과 전체 일치 검사 주기를 포함합니다.

연습 문제

버전 4 취소 이벤트(게시 건수=0, 조회수=0)를 넣고 처리하세요. 일별 행이 없어지고 같은 취소 이벤트를 다시 삽입하려 하면 uq_stat_event_version에서 거절되어야 합니다.

해설과 예시 답안

취소도 최신 절대 기여값입니다. 처리기는 게시글 행을 0으로 갱신한 뒤 영향받은 날짜를 지우고 HAVING 조건 때문에 새 일별 행을 만들지 않습니다.

INSERT INTO post_change_event
 (post_id,source_version,author_id,stat_date,published_count,total_views,produced_at)
VALUES(@post_id,4,@author_id,'2026-07-16',0,0,'2026-07-16 04:00:00');
CALL process_post_stats_microbatch('daily-stat-v1',100);
SELECT * FROM daily_post_stats_incremental
WHERE author_id=@author_id AND stat_date='2026-07-16';
INSERT INTO post_change_event
 (post_id,source_version,author_id,stat_date,published_count,total_views,produced_at)
VALUES(@post_id,4,@author_id,'2026-07-16',0,0,'2026-07-16 04:01:00');
-- ERROR 1062: uq_stat_event_version

첫 SELECT는 0행입니다. 기여값에는 source_version 4와 조회수 0의 삭제 표식이 남습니다. 중복 버전 INSERT는 대상을 건드리기 전에 유일 제약 오류로 끝납니다.

핵심 정리

증분 이벤트를 합계에 직접 더하면 재소비와 정정에서 불일치가 발생합니다.
게시글별 최신 기여값은 게시 건수·조회수의 절대 상태를 보존합니다.
이전/신규 변경된 키를 다시 SUM하고 체크포인트까지 한 트랜잭션으로 커밋합니다.
마이크로배치 크기와 재호출 횟수는 최종 대상 값을 바꾸지 않아야 합니다.

다음 장에서는 서로 다른 게시글 타입을 관계형 스키마에 표현하는 상속 전략과 EAV 경계를 비교합니다.