같은 데이터, 다른 처리 - Spring Batch로 주간/월간 랭킹 집계하기

진짜?

tl;dr - 지난 포스트에서 Redis ZSET 기반 일간 실시간 랭킹을 구축했습니다. 주간/월간도 같은 방식으로 하면 될 줄 알았는데, 비선형 스코어(log1p)가 발목을 잡았습니다. ZINCRBY로 7일치를 누적하면 취소 시 역산이 불가능합니다. 결국 Spring Batch로 원본 데이터에서 처음부터 다시 계산하는 배치 집계를 도입했고, Materialized View(MV) 테이블에 TOP 100을 적재하는 구조로 해결했습니다.

 

 

---

1. ZINCRBY로 주간 랭킹을 만들 수 없는 이유

이전 포스트에서 작성한 일간 랭킹은 잘 동작하고 있었습니다. product_daily_metrics 테이블에 일간 집계 데이터가 쌓이고, 5분 주기 스케줄러가 이걸 읽어서 Redis ZSET에 반영합니다. 조회는 ZREVRANGE로 O(log N)에 끝납니다.

 

주간 랭킹도 같은 방식으로 하면 되지 않을까요? 월요일부터 일요일까지 7일치 이벤트를 ZINCRBY로 누적하면 주간 ZSET이 완성됩니다.

문제는 스코어 공식에 log1p(orderAmount)가 있다는 겁니다.

 

주문 10,000원 → log1p(10000) ≈ 9.21 → ZINCRBY +9.21
주문 취소    → ZINCRBY -???  ← 이전 총 금액을 모르니 역산 불가

log1p는 비선형 함수입니다. log1p(a + b) ≠ log1p(a) + log1p(b)이므로, 특정 주문의 기여분만 빼는 게 수학적으로 불가능합니다. ZSET에는 합산된 score만 저장되어 있으니 원래 금액을 역추적할 방법이 없습니다.

 

선형 공식(views × 0.1 + likes × 0.2)이라면 ZINCRBY로 증감이 자유롭습니다. 하지만 주문 금액의 스케일을 눌러주기 위해 log1p를 쓴 이상, 증분 방식은 포기해야 합니다.

 

다른 선택지도 검토했습니다.

선택지	문제
API 요청 시 product_daily_metrics에서 실시간 집계	매 요청마다 7~30일 × 전체 상품 GROUP BY + ORDER BY. 읽기 부하가 상품 수에 비례
Redis ZINCRBY로 주간/월간 ZSET 유지	비선형 스코어에서 취소/수정 시 역산 불가
배치로 주기적 집계 → MV 테이블 적재	원본에서 처음부터 다시 계산하므로 취소/수정이 자동 반영

 

여기서 저는 배치를 선택했습니다. 원본 데이터(product_daily_metrics)에서 매번 처음부터 계산하면, 중간에 주문이 취소되었든 수정되었든 최종 상태가 자동으로 반영됩니다. Volume 9에서 DB를 SSOT로 마이그레이션한 덕분에 배치 Reader의 소스 테이블이 이미 준비되어 있었습니다.

 

---

2. Chunk-Oriented 구조 - Reader, Processor, Writer

Spring Batch의 Chunk-Oriented 패턴을 적용했습니다. TOP 100만 적재하는 작업이라 Tasklet 하나로도 충분하지만, Reader → Processor → Writer 단계 분리를 학습하기 위해 Chunk-Oriented를 선택했습니다.

 

Reader: 집계 + TOP 100 필터링

Reader는 JdbcCursorItemReader로 product_daily_metrics에서 기간별 SUM 집계를 수행합니다.

SELECT
    product_id,
    SUM(view_count)  AS total_views,
    SUM(like_count)  AS total_likes,
    SUM(order_amount) AS total_amount
FROM product_daily_metrics
WHERE metric_date BETWEEN ? AND ?
GROUP BY product_id
ORDER BY (SUM(view_count) * 0.1 + SUM(like_count) * 0.2
        + LOG(1 + SUM(order_amount)) * 0.7) DESC, product_id ASC
LIMIT 100

ORDER BY에 product_id ASC를 tie-breaker로 추가했습니다. 스코어가 동일한 상품이 있으면 MySQL이 비결정적 순서로 반환하는데, 배치를 재실행할 때마다 ranking이 바뀌면 안 되니까요.

 

Reader의 출력은 AggregatedMetricRow record입니다.

public record AggregatedMetricRow(
        long productId,
        long viewCount,
        long likeCount,
        long orderAmount
) {}

 

Processor: 스코어 계산 + ranking 부여

Processor는 RankingScoreProcessor로 가중치 기반 스코어를 계산하고, AtomicInteger로 순서대로 ranking을 부여합니다.

public class RankingScoreProcessor implements ItemProcessor<AggregatedMetricRow, RankingScoreRow> {

    private final AtomicInteger rankCounter = new AtomicInteger(0);

    @Override
    public RankingScoreRow process(AggregatedMetricRow item) {
        double score = item.viewCount() * 0.1
                + item.likeCount() * 0.2
                + Math.log1p(item.orderAmount()) * 0.7;

        return new RankingScoreRow(
                item.productId(),
                item.viewCount(),
                item.likeCount(),
                item.orderAmount(),
                score,
                rankCounter.incrementAndGet()
        );
    }
}

여기서 눈에 띄는 게 하나 있습니다. 스코어 공식이 Reader SQL과 Processor 양쪽에 존재합니다. Reader의 ORDER BY와 Processor의 Math.log1p() 계산이 동일한 공식입니다. DRY 위반처럼 보이지만, 목적이 다릅니다.

 

• Reader: "누구를 뽑을지" - TOP 100 필터링용
• Processor: "뽑힌 데이터에 무엇을 부여할지" - 저장할 정확한 score 값 산출

 

Reader SQL에서 ROW_NUMBER()까지 처리하고 Processor를 제거하면 공식이 한 곳으로 모입니다. 하지만 그러면 Chunk-Oriented의 단계 분리 원칙과 충돌합니다. 이 트레이드오프는 의식적으로 허용했습니다.

 

Writer: DELETE + INSERT

Writer는 해당 기간의 기존 행을 삭제하고, 새로운 TOP 100을 삽입합니다.

return items -> {
    jdbcTemplate.update(
            "DELETE FROM mv_product_rank_weekly WHERE ranking_week = :yearWeek",
            new MapSqlParameterSource("yearWeek", yearWeek)
    );

    String sql = """
            INSERT INTO mv_product_rank_weekly
                (product_id, ranking_week, view_count, like_count,
                 order_amount, score, ranking, updated_at)
            VALUES
                (:productId, :yearWeek, :viewCount, :likeCount,
                 :orderAmount, :score, :ranking, :updatedAt)
            """;

    SqlParameterSource[] batchParams = items.getItems().stream()
            .map(row -> new MapSqlParameterSource()
                    .addValue("productId", row.productId())
                    // ... 나머지 파라미터
            ).toArray(SqlParameterSource[]::new);
    jdbcTemplate.batchUpdate(sql, batchParams);
};

처음에는 ON DUPLICATE KEY UPDATE(UPSERT)로 구현했습니다. 같은 (product_id, ranking_week) 조합이 있으면 갱신하고, 없으면 삽입하는 방식입니다. 멱등성도 보장되고 깔끔해 보였습니다.

 

문제를 발견한 건 이런 시나리오에서였습니다.

1차 실행: 상품 A, B, C가 TOP 3 → MV에 A, B, C 저장
2차 실행: 상품 C가 탈락, A, B만 TOP 2 → UPSERT로 A, B만 갱신
결과: MV에 A(갱신), B(갱신), C(그대로 남아있음) ← stale 행

UPSERT는 "존재하는 행을 갱신"하는 것이지 "더 이상 존재하지 않는 행을 제거"하는 게 아닙니다. 탈락한 상품 C가 유령처럼 MV에 남아 있게 됩니다.

DELETE + INSERT로 변경했습니다. 해당 기간의 행을 전부 지우고 새로 넣으니까 stale 행이 남을 수 없습니다. 두 연산이 같은 트랜잭션(Chunk의 트랜잭션) 안에서 실행되므로 원자적입니다.

 

 

---

3. MV 테이블 설계 - 복합 PK

집계 결과를 저장하는 MV(Materialized View) 테이블을 주간/월간 각각 만들었습니다.

mv_product_rank_weekly
  PK: (product_id, ranking_week)   -- "2026-W15"
  INDEX: (ranking_week, ranking)   -- 조회 최적화

mv_product_rank_monthly
  PK: (product_id, ranking_month)  -- "2026-04"
  INDEX: (ranking_month, ranking)

컬럼명을 year_month로 설계했다가 MySQL 예약어(YEAR_MONTH은 INTERVAL keyword)와 충돌해서 Hibernate DDL 생성 시 syntax error가 났습니다. backtick으로 이스케이프하면 당장은 동작하지만, 근본적으로 ranking_month/ranking_week로 이름을 바꿔서 해결했습니다.

 

 

---

4. API 데이터 소스 분기 - Facade 패턴

일간은 Redis ZSET, 주간/월간은 DB MV 테이블입니다. 데이터 소스가 다른데 어디서 분기할지가 문제였습니다.

선택지	문제
Controller에서 분기	Controller가 데이터 소스를 알게 됨
AppService에서 분기	일간은 RankingAppService(Redis), 주간/월간은 MvRankingAppService(DB)로 이미 분리됨. 하나가 둘 다 아는 건 책임 과잉
Facade에서 분기	period에 따라 적절한 AppService를 호출하고 상품 정보를 Aggregation

 

RankingFacade가 period 파라미터에 따라 적절한 AppService를 호출하는 구조로 결정했습니다. Facade의 원래 역할이 여러 AppService를 오케스트레이션하는 것이니까 자연스럽습니다.

 

Controller(period=weekly)
  → RankingFacade.getWeeklyTopRankings()
    → MvRankingAppService.getWeeklyRankings()  -- DB MV 조회
    → ProductAppService.getByIds()              -- 상품 정보
    → enrichWithProductInfo()                   -- 조합

기존 일간 랭킹의 enrichWithProductInfo() 메서드를 주간/월간에서도 그대로 재사용합니다. MvRankingAppService가 MV 조회 결과를 RankingEntry record로 변환하면, 이후 파이프라인은 일간과 동일합니다.

 

 

---

5. V9 → V10 연결 - DB SSOT가 배치를 쉽게 만들었다

이번 배치 구현이 수월했던 건 이전의 설계 판단 덕분입니다.

 

저번 포스트에서 랭킹 시스템의 데이터 소스를 Redis에서 DB로 옮긴 적이 있습니다. Redis HINCRBY로 집계하던 걸 product_daily_metrics 테이블에 UPSERT하는 방식으로 바꿨습니다. 당시에는 멱등성과 영속성 확보가 목적이었는데, 이 결정이 이번에 빛을 발했습니다.

 

[V9에서 만든 구조]
Kafka Consumer → DB: product_daily_metrics (SSOT)
스케줄러 (5분 주기) → Redis ZSET (읽기 캐시)

[V10에서 활용]
Spring Batch Reader → DB: product_daily_metrics (동일 테이블!)
Spring Batch Writer → DB: mv_product_rank_weekly / monthly

만약 Redis를 SSOT로 유지했다면 배치 Reader가 Redis에서 데이터를 뽑아야 했습니다. Redis ZSET에서 기간별 필터링 + GROUP BY를 하는 건 사실상 불가능하니, 별도 집계 로직을 만들어야 했을 겁니다.

 

"DB가 SSOT, Redis는 읽기 캐시" 판단이 "배치 Reader는 DB에서 SQL로 집계" 구조를 자연스럽게 연결해줬습니다. 이전의 설계 판단이 다음의 구현 난이도를 결정한다는 걸 체감한 부분입니다.

 

 

---

 

6. 테스트 - 멱등성, tie-break, stale 행 제거

배치 E2E 테스트를 11건 작성했습니다. Testcontainers로 MySQL을 띄우고, @SpringBatchTest로 Job을 실행합니다.

 

멱등성 검증

동일 기간에 배치를 두 번 실행해도 결과가 중복되지 않아야 합니다.

@Test
void monthlyRankingJob_idempotent() throws Exception {
    insertMetrics(1L, LocalDate.of(2026, 1, 1), 100, 50, 10000);

    // 1차 실행
    jobLauncherTestUtils.launchJob(jobParameters1);
    // 2차 실행
    JobExecution jobExecution = jobLauncherTestUtils.launchJob(jobParameters2);

    List<Map<String, Object>> results = jdbcTemplate.queryForList(
            "SELECT * FROM mv_product_rank_monthly WHERE ranking_month = '2026-01'"
    );
    assertThat(results).hasSize(1);  // 중복 없이 1건만 존재
}

DELETE + INSERT 방식이니까 2차 실행 시 기존 행을 지우고 다시 넣습니다. RunIdIncrementer로 같은 파라미터로도 재실행이 가능합니다.

 

tie-break 검증

동일 점수의 상품 3개를 넣고, product_id 오름차순으로 정렬되는지 확인합니다.

@Test
void monthlyRankingJob_tieBreakByProductId() throws Exception {
    insertMetrics(99L, LocalDate.of(2026, 5, 1), 100, 50, 10000);
    insertMetrics(11L, LocalDate.of(2026, 5, 1), 100, 50, 10000);
    insertMetrics(55L, LocalDate.of(2026, 5, 1), 100, 50, 10000);

    jobLauncherTestUtils.launchJob(jobParameters);

    List<Map<String, Object>> results = jdbcTemplate.queryForList(
            "SELECT * FROM mv_product_rank_monthly WHERE ranking_month = '2026-05' ORDER BY ranking"
    );
    assertAll(
            () -> assertThat(results.get(0).get("product_id")).isEqualTo(11L),
            () -> assertThat(results.get(1).get("product_id")).isEqualTo(55L),
            () -> assertThat(results.get(2).get("product_id")).isEqualTo(99L)
    );
}

ORDER BY ... DESC, product_id ASC 덕분에 매번 동일한 순서가 보장됩니다.

 

stale 행 제거 검증

1차 실행에서 상품 A, B가 적재되고, 2차 실행에서 A만 남았을 때 B가 MV에서 사라지는지 확인합니다. UPSERT 방식이었으면 이 테스트가 실패합니다.

@Test
void monthlyRankingJob_removesStaleEntries() throws Exception {
    // 1차: 상품 A, B
    insertMetrics(1L, LocalDate.of(2026, 6, 1), 100, 50, 10000);
    insertMetrics(2L, LocalDate.of(2026, 6, 1), 80, 40, 8000);
    jobLauncherTestUtils.launchJob(jobParameters1);
    assertThat(firstRun).hasSize(2);

    // 2차: 상품 A만
    jdbcTemplate.execute("TRUNCATE TABLE product_daily_metrics");
    insertMetrics(1L, LocalDate.of(2026, 6, 1), 100, 50, 10000);
    jobLauncherTestUtils.launchJob(jobParameters2);

    assertThat(secondRun).hasSize(1);  // B가 제거됨
    assertThat(secondRun.get(0).get("product_id")).isEqualTo(1L);
}

 

 

---

7. 10주를 돌아보며

이커머스라는 도메인을 바탕으로 약 10주간 테크니컬 라이팅을 작성해왔는데요. 지난 10주를 한 줄로 요약하면 이렇습니다.

 

주차	테마	한 줄 요약
1	Member VO	비밀번호 하나에도 도메인 규칙이 있다
2	설계 문서	ERD와 시퀀스 다이어그램으로 머릿속 그림을 검증했다
3	DDD 4계층	Domain → Application → Infrastructure → Interfaces로 전체 커머스 플랫폼 구축
4	동시성 제어	"돌아가는 코드"와 "동시에 돌려도 안전한 코드"는 다르다
5	캐시 + 인덱스	1,000만 건에서 2,717배 개선, Redis Cache-Aside 패턴
6	PG 결제	외부 시스템은 내부와 다른 방식으로 실패한다
7	이벤트 파이프라인	Outbox + Kafka로 "DB 커밋은 됐는데 메시지는 안 간" 문제 해결
8	Redis 대기열	Thundering Herd를 배치 크기 산정 + Jitter로 2단계 제어
9	실시간 랭킹	double 하나에 메인 점수 + 타이브레이커를 인코딩
10	배치 랭킹	같은 데이터도 용도에 따라 처리 방식이 달라야 한다

 

나열하면 10개의 독립된 과제처럼 보이지만, 돌아보면 연결 고리가 있습니다.

 

3주차에 만든 주문 도메인은 4주차에서 비관적 락으로 보호받았고, 6주차에서 PG 결제와 연결됐습니다. 7주차에서 주문 이벤트가 Kafka로 나갔고, commerce-streamer가 소비하면서 product_metrics 테이블에 적재됐습니다. 9주차에서 이 집계 데이터를 Redis ZSET으로 실시간 랭킹에 반영했고, DB SSOT 마이그레이션으로 만든 product_daily_metrics가 10주차 배치 Reader의 소스가 됐습니다.

 

[V3 주문] → [V4 재고 락] → [V6 PG 결제] → [V7 Kafka 이벤트]
                                                    ↓
		[V10 배치 MV] ← [V9 일간 집계] ← [V7 streamer 적재]
                                                    ↓
                                    [V8 대기열] → [V5 캐시 조회]

처음에는 "이번 주 과제"를 치우는 데 급급했는데, 7주차쯤부터 이전 주의 산출물이 이번 주의 입력이 되는 감각이 생겼습니다. 가장 많이 배운 건 두 가지입니다.

 

첫째, 같은 "랭킹"이라는 기능인데 처리 방식이 다릅니다. 일간은 실시간(5분 주기 스케줄러 + Redis ZSET), 주간/월간은 배치(Spring Batch + MV 테이블). 데이터의 갱신 주기와 스코어 공식의 특성에 따라 적절한 도구가 다릅니다. 이 감각은 5주차 캐시 무효화에서 처음 생겼습니다. Admin 상품 수정은 즉시 evict, 좋아요 토글은 TTL 만료에 맡기는 식으로 이벤트 성격에 따라 전략을 달리했던 경험이 시작이었습니다.

 

둘째, 이전 주차의 설계 판단이 다음 주차의 난이도를 결정합니다. V9에서 DB SSOT를 선택하지 않았다면 V10 배치 구현이 훨씬 복잡했을 겁니다. V7에서 Outbox 패턴을 도입하지 않았다면 V9에서 이벤트 기반 집계가 불가능했을 겁니다. 당장은 복잡해 보이는 판단이 나중에 선택지를 넓혀주는 경우가 있었습니다. 오버엔지니어링이라 생각하기보다는, 다음 설계의 확장성을 미리 대비하는 것이 좋은 것 같습니다.

 

 

---

8. P.S.

가중치 0.7의 근거

스코어 공식은 view × 0.1 + like × 0.2 + log1p(amount) × 0.7입니다. V9 일간 랭킹에서는 주문 가중치가 0.6이었는데, V10 배치에서는 0.7로 올렸습니다. V9에서 carry-over(전일 점수 계승)에 0.1을 배분했지만, 배치에서는 carry-over가 없으므로 그 0.1을 주문에 재배분했습니다. 합계 1.0은 유지됩니다.

 

Chunk 크기와 DELETE 타이밍

현재 CHUNK_SIZE = 100이고 LIMIT 100이라 Writer가 1번만 호출됩니다. DELETE와 INSERT가 한 청크 트랜잭션 안에 묶이니까 원자적입니다. 만약 TOP N이 chunk 크기를 넘어가면 첫 번째 청크에서 DELETE가 실행되고, 두 번째 청크의 INSERT가 완료되기 전까지 데이터가 불완전한 구간이 생깁니다. 현재 TOP 100 규모에서는 문제가 없지만, 확장 시 고려해야 할 포인트입니다.

 

MySQL LOG vs Java Math.log1p

Reader SQL의 LOG(1 + SUM(order_amount))와 Processor의 Math.log1p(orderAmount)는 같은 계산입니다. MySQL에 log1p 함수가 없어서 LOG(1 + x)로 쓴 것뿐입니다. 둘 다 자연로그 ln(1 + x)이고, IEEE 754 double 범위에서 동일한 값을 반환합니다.