数据归档

ADX_REPORT_BACKFILL.md

@@ -75,7 +75,8 @@ adx:
 
 - 汇总源数据来自三张热表，不直接扫月历史表
 - 统计窗口不包含当天数据，只统计到昨天 24:00 为止
-- 每次任务都会重算 lookback 窗口内的数据，再通过 upsert 覆盖旧值
+- 归档任务每 5 分钟跑一次，但 daily stats 默认每天按上海时区 `01:00`、`10:00`、`17:00` 跑三次
+- 每次 daily stats 执行时，会重算 lookback 窗口内的数据，再通过 upsert 覆盖旧值
 - 默认只回算近 7 个自然日，这样不会和“30 天前数据归档”窗口重叠
 - 即使配置误设得过大，代码也会自动限制为 `retentionDays - 1`，避免边界日期统计失真
 - 所以近 7 个自然日如果有补写、重复清理、延迟写入，daily stats 会被自动修正
@@ -84,6 +85,7 @@ adx:
 
 ```yaml
 adx:
+  ad-bid-report-stats-run-times: ["01:00", "10:00", "17:00"]
   ad-bid-report-stats-lookback-days: 7
 ```
 
@@ -114,5 +116,6 @@ adx:
   ad-bid-report-retention-days: 30
   ad-bid-report-archive-batch-size: 5000
   ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-run-times: ["01:00", "10:00", "17:00"]
   ad-bid-report-stats-lookback-days: 7
 ```

src/main/java/com/adx/tencent/config/AppProperties.java

@@ -105,6 +105,7 @@ public class AppProperties {
     private int adBidReportRetentionDays = 30;
     private int adBidReportArchiveBatchSize = 5000;
     private int adBidReportArchiveMaxBatchesPerRun = 20;
+    private List<String> adBidReportStatsRunTimes = List.of("01:00", "10:00", "17:00");
     private int adBidReportStatsLookbackDays = 7;
 
     // Task Lock
@@ -729,6 +730,14 @@ public class AppProperties {
         this.adBidReportArchiveMaxBatchesPerRun = v;
     }
 
+    public List<String> getAdBidReportStatsRunTimes() {
+        return adBidReportStatsRunTimes;
+    }
+
+    public void setAdBidReportStatsRunTimes(List<String> v) {
+        this.adBidReportStatsRunTimes = v;
+    }
+
     public int getAdBidReportStatsLookbackDays() {
         return adBidReportStatsLookbackDays;
     }

src/main/java/com/adx/tencent/report/AdBidReportBackgroundTasks.java

@@ -10,6 +10,14 @@ import org.springframework.boot.ApplicationRunner;
 import org.springframework.lang.Nullable;
 import org.springframework.stereotype.Component;
 
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -18,6 +26,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class AdBidReportBackgroundTasks implements ApplicationRunner {
 
     private static final Logger log = LoggerFactory.getLogger(AdBidReportBackgroundTasks.class);
+    private static final ZoneId REPORT_ZONE = ZoneId.of("Asia/Shanghai");
+    private static final List<LocalTime> DEFAULT_STATS_RUN_TIMES = List.of(
+            LocalTime.of(1, 0),
+            LocalTime.of(10, 0),
+            LocalTime.of(17, 0)
+    );
 
     @Autowired private AppProperties props;
     @Autowired(required = false) private AdBidReportStore adBidReportStore;
@@ -50,11 +64,23 @@ public class AdBidReportBackgroundTasks implements ApplicationRunner {
 
     private void runReportJob(LeaderElection.StopSignal stopSignal) {
         long intervalMs = props.getAdBidReportInterval().toMillis();
+        List<LocalTime> statsRunTimes = resolveStatsRunTimes(props.getAdBidReportStatsRunTimes());
+        String lastStatsSlotKey = null;
         while (!stopSignal.isStopped()) {
             try {
-                AdBidReportStore.RunSummary summary = adBidReportStore.runOnce();
-                log.info("[AdBidReport] done | archivedRows={} | statsRows={}",
-                        summary.archivedRows, summary.statsRows);
+                ZonedDateTime now = ZonedDateTime.now(REPORT_ZONE);
+                String currentStatsSlotKey = currentStatsSlotKey(now, statsRunTimes);
+                boolean includeStats = currentStatsSlotKey != null && !currentStatsSlotKey.equals(lastStatsSlotKey);
+                AdBidReportStore.RunSummary summary = adBidReportStore.runOnce(includeStats);
+                if (summary.statsRun) {
+                    lastStatsSlotKey = currentStatsSlotKey;
+                }
+                ZonedDateTime nextStatsAt = nextStatsRunAt(now, statsRunTimes);
+                log.info("[AdBidReport] done | archivedRows={} | statsRows={} | statsRun={} | nextStatsAt={}",
+                        summary.archivedRows,
+                        summary.statsRows,
+                        summary.statsRun,
+                        nextStatsAt == null ? "-" : nextStatsAt);
             } catch (Exception e) {
                 log.error("[AdBidReport] error: {}", e.getMessage(), e);
             }
@@ -78,4 +104,51 @@ public class AdBidReportBackgroundTasks implements ApplicationRunner {
             Thread.currentThread().interrupt();
         }
     }
+
+    private static List<LocalTime> resolveStatsRunTimes(List<String> configured) {
+        List<LocalTime> parsed = new ArrayList<>();
+        if (configured != null) {
+            for (String value : configured) {
+                if (value == null || value.isBlank()) {
+                    continue;
+                }
+                try {
+                    parsed.add(LocalTime.parse(value.trim()));
+                } catch (DateTimeParseException e) {
+                    log.warn("[AdBidReport] ignore invalid stats run time: {}", value);
+                }
+            }
+        }
+        if (parsed.isEmpty()) {
+            parsed.addAll(DEFAULT_STATS_RUN_TIMES);
+        }
+        parsed.sort(Comparator.naturalOrder());
+        return parsed.stream().distinct().toList();
+    }
+
+    private static String currentStatsSlotKey(ZonedDateTime now, List<LocalTime> statsRunTimes) {
+        LocalDate date = now.toLocalDate();
+        LocalTime time = now.toLocalTime();
+        LocalTime matched = null;
+        for (LocalTime candidate : statsRunTimes) {
+            if (!candidate.isAfter(time)) {
+                matched = candidate;
+            }
+        }
+        if (matched == null) {
+            return null;
+        }
+        return date + "T" + matched;
+    }
+
+    private static ZonedDateTime nextStatsRunAt(ZonedDateTime now, List<LocalTime> statsRunTimes) {
+        LocalDate date = now.toLocalDate();
+        LocalTime time = now.toLocalTime();
+        for (LocalTime candidate : statsRunTimes) {
+            if (candidate.isAfter(time)) {
+                return date.atTime(candidate).atZone(REPORT_ZONE);
+            }
+        }
+        return date.plusDays(1).atTime(statsRunTimes.get(0)).atZone(REPORT_ZONE);
+    }
 }

src/main/java/com/adx/tencent/report/AdBidReportStore.java

@@ -58,13 +58,20 @@ public class AdBidReportStore {
     }
 
     public RunSummary runOnce() {
+        return runOnce(true);
+    }
+
+    public RunSummary runOnce(boolean includeStats) {
         RunSummary summary = new RunSummary();
         Timestamp cutoff = Timestamp.from(Instant.now().minusSeconds(retentionDays * 24L * 3600L));
         try (Connection conn = dataSource.getConnection()) {
             conn.setAutoCommit(false);
             for (MediaTable media : MediaTable.values()) {
                 summary.archivedRows += archiveMedia(conn, media, cutoff);
-                summary.statsRows += rollupStats(conn, media);
+                if (includeStats) {
+                    summary.statsRows += rollupStats(conn, media);
+                    summary.statsRun = true;
+                }
             }
             conn.commit();
             return summary;
@@ -289,6 +296,7 @@ public class AdBidReportStore {
     public static class RunSummary {
         public long archivedRows;
         public long statsRows;
+        public boolean statsRun;
     }
 
     private enum MediaTable {

src/main/resources/application-dev.yml

@@ -67,6 +67,7 @@ adx:
   ad-bid-report-retention-days: 30
   ad-bid-report-archive-batch-size: 5000
   ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-run-times: ["01:00", "10:00", "17:00"]
   ad-bid-report-stats-lookback-days: 7
 
   # --- Task ---

src/main/resources/application-prod.yml

@@ -69,6 +69,7 @@ adx:
   ad-bid-report-retention-days: 30
   ad-bid-report-archive-batch-size: 5000
   ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-run-times: ["01:00", "10:00", "17:00"]
   ad-bid-report-stats-lookback-days: 7
 
   # --- Task ---

src/main/resources/application-test.yml

@@ -68,6 +68,7 @@ adx:
   ad-bid-report-retention-days: 30
   ad-bid-report-archive-batch-size: 5000
   ad-bid-report-archive-max-batches-per-run: 20
+  ad-bid-report-stats-run-times: ["01:00", "10:00", "17:00"]
   ad-bid-report-stats-lookback-days: 7
 
   # --- Task ---