年度流量选品-新增asin_bought_month 最新月份对应月销

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -3,22 +3,13 @@
    @Description : 月流量选品 Doris 落地脚本，同时关联产出年度流量选品数据
                   流程：
                     [Step 1] Doris 建表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
-                    [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化（与 dwt.us_flow_asin_30day
-                             DDL 对齐），合并为一步
-                    [Step 3] 写入 Doris dwt 主表 dwt.{site}_flow_asin_month
-                    [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day：强制重写当月数据 +
-                             补齐缺失月份，靠 UNIQUE KEY+MOW+sequence_col(date_info) 自动去重只
-                             保留每个 asin 最新一条快照，并删除近12月窗口之外的过期数据
-                    [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive
-                             dwt_flow_asin_year（由 dwt/dwt_flow_asin_year.py 算好写入）对应
-                             site_name+date_info 分区同步过来，TRUNCATE 后全量重算
-                    [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表
-                             selection.{site}_flow_asin_month_{yyyy_mm}[_test]
-                    [Step 7] Doris INSERT 到 selection 年表
-                             selection.{site}_flow_asin_365day[_test]
+                    [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化（与 dwt.us_flow_asin_30day DDL 对齐）
+                    [Step 3] 写入 Doris dwt 月表 dwt.{site}_flow_asin_month
+                    [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day，并删除近12月窗口之外的过期数据
+                    [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive dwt_flow_asin_year 同步过来
+                    [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
+                    [Step 7] Doris INSERT 到 selection 年表 selection.{site}_flow_asin_365day[_test]
                     [Step 8] 更新 MySQL workflow_everyday 流程记录表（月/年各写一条，仅 formal 模式）
-                  依赖：dwt/dwt_flow_asin_year.py 必须已经跑完同一个 site_name+date_info，
-                       Step 5 才能同步到有效的年度聚合数据
                   支持 us / uk / de 三站点
                   支持 formal / test 模式：
                     - formal：selection 表名无后缀，更新流程记录表
@@ -1028,7 +1019,7 @@ WHERE f.date_info = '{date_info}'
 def build_year_create_table_sql(table_name):
     """构建 selection.{table_name} 建表语句：以 selection 月表（build_month_create_table_sql）
     为基准调整而来——
-      剔除月度专属字段：asin_bought_month / asin_source_flag / bsr_last_seen_at /
+      剔除月度专属字段：asin_source_flag / bsr_last_seen_at /
                        bsr_seen_count_30d / nsr_last_seen_at / nsr_seen_count_30d
       末尾追加：latest_date_info + dwt.{site}_flow_asin_365day_extra 里的年度聚合字段
     旧表已手动清理，直接 CREATE 新表，不再需要 ALTER 兼容旧表结构"""
@@ -1059,6 +1050,7 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `low_star`                    INT              NULL,
     `bsr_orders`                  INT              NULL,
     `bsr_orders_sale`             DECIMAL(20,2)    NULL,
+    `asin_bought_month`           INT              NULL,
     `ao_val`                      DECIMAL(20,4)    NULL,
     `zr_counts`                   INT              NULL,
     `sp_counts`                   INT              NULL,
@@ -1236,6 +1228,7 @@ SELECT
     f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
     f.bsr_orders,
     f.bsr_orders_sale,
+    f.asin_bought_month,
     f.ao_val,
     f.zr_counts,
     f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
@@ -1491,8 +1484,7 @@ def write_year_table_batched(site_name, live_table):
         f"ALTER TABLE `selection`.`{live_table}` REPLACE WITH TABLE `{copy_table}` "
         f"PROPERTIES('swap' = 'false')"
     ])
-    print(f"[Step 7] selection.{live_table} 已切换为最新数据，"
-          f"selection.{copy_table} 保留的是交换前的旧数据（可用于回滚）")
+    print(f"[Step 7] selection.{live_table} 已切换为最新数据")
 
 
 # ============================================================