流量选品月-增加keepa类型指标

Pyspark_job/doris_handle/dws_flow_asin_refresh.py

@@ -4,17 +4,18 @@
                   统一只处理最新一个月，且全部改成"发生变化才更新"的部分列更新，减少 Doris 压力
                   【背景】
                   selection.{site}_flow_asin_month_{yyyy_mm}(月表) / selection.{site}_flow_asin_365day(年表)
-                  这两张物化表里，只有以下 9 个字段依赖 dwd_asin_profit_rate_latest(利润率) /
+                  这两张物化表里，只有以下字段依赖 dwd_asin_profit_rate_latest(利润率) /
                   dwd_keepa_asin_detail(Keepa) 这两张每日刷新的源表，其余几十个字段都是月度/年度批跑时
                   算好、不需要每天动的：
                       ocean_profit, air_profit,
                       launch_time, launch_time_type,
                       tracking_since, tracking_since_type,
-                      package_length, package_width, package_height, item_weight
+                      package_length, package_width, package_height, item_weight,
+                      keepa_type, keepa_item_type, keepa_type_cn, keepa_item_type_cn
                   月表/年表都是 UNIQUE KEY(asin) + enable_unique_key_merge_on_write=true，支持部分列更新
-                  （INSERT INTO table(asin, 这9个字段) SELECT ...），不需要整表/整分区重算。
+                  （INSERT INTO table(asin, 这批字段) SELECT ...），不需要整表/整分区重算。
                   【4 个子任务，main() 里按顺序执行】
-                  1) 刷新月表：selection.{site}_flow_asin_month_{最新月}，只更新这9个字段
+                  1) 刷新月表：selection.{site}_flow_asin_month_{最新月}，只更新这批字段
                   2) 刷新年表：selection.{site}_flow_asin_365day，限定 latest_date_info=最新月 这批 asin
                   3) 刷新利润率趋势中间表：dwt.dwt_asin_profit_rate_history，只处理最新1个月（原来是近3月）
                   4) Spark 聚合中间表全量历史 → 趋势物理表 selection.{site}_asin_profit_rate_trend
@@ -52,6 +53,7 @@ PATTERN_REFRESH_FIELDS = {
     'launch_time', 'launch_time_type',
     'tracking_since', 'tracking_since_type',
     'package_length', 'package_width', 'package_height', 'item_weight',
+    'keepa_type', 'keepa_item_type', 'keepa_type_cn', 'keepa_item_type_cn',
 }
 
 
@@ -106,10 +108,10 @@ def get_latest_month(spark):
     return str(max_date_info)
 
 
-# ===== 任务1/2 共用：月表 + 年表的"9个每日变化字段"部分列更新 =====
+# ===== 任务1/2 共用：月表 + 年表的"每日变化字段"部分列更新 =====
 
 def _daily_field_exprs():
-    """9个每日变化字段各自的"新值"表达式（与 dwt_flow_asin_month.py 里的口径保持一致），
+    """每日变化字段各自的"新值"表达式（与 dwt_flow_asin_month.py 里的口径保持一致），
     SELECT 取新值 和 WHERE 判断是否变化 两处都要用到，这里只定义一次"""
     launch_time_expr = "COALESCE(f.launch_time, kp.keepa_launch_time)"
     launch_time_type_expr = f"""CASE
@@ -147,12 +149,16 @@ def _daily_field_exprs():
         ("package_width", "kp.package_width"),
         ("package_height", "kp.package_height"),
         ("item_weight", item_weight_expr),
+        ("keepa_type", "kp.keepa_type"),
+        ("keepa_item_type", "kp.keepa_item_type"),
+        ("keepa_type_cn", "kp.keepa_type_cn"),
+        ("keepa_item_type_cn", "kp.keepa_item_type_cn"),
     ]
 
 
 def build_daily_fields_refresh_sql(site_name, table_name, extra_where_sql=None):
     """构建月表/年表共用的部分列更新 SQL：
-    只更新9个每日变化字段，且只对"新值跟表里已存的值不一致"的行才写入
+    只更新每日变化字段，且只对"新值跟表里已存的值不一致"的行才写入
     （<=> 是 NULL-safe 比较，已在生产 Doris 上验证过 有值<=>NULL 会判定为不相等）
     extra_where_sql: 年表需要额外限定 latest_date_info=最新月，月表不需要（整表就是那一个月）
     """
@@ -182,7 +188,7 @@ def build_daily_fields_refresh_sql(site_name, table_name, extra_where_sql=None):
 # ===== 任务1：刷新月表 =====
 
 def refresh_month_table(site_name, latest_month):
-    """任务1: 刷新 selection.{site}_flow_asin_month_{最新月} 的9个每日变化字段（表已由月度主流程建好）"""
+    """任务1: 刷新 selection.{site}_flow_asin_month_{最新月} 的每日变化字段（表已由月度主流程建好）"""
     table_name = f"{site_name}_flow_asin_month_{latest_month.replace('-', '_')}"
     print(f"\n========== [任务1/月表] selection.{table_name} ==========")
     sql = build_daily_fields_refresh_sql(site_name, table_name)
@@ -193,7 +199,7 @@ def refresh_month_table(site_name, latest_month):
 # ===== 任务2：刷新年表 =====
 
 def refresh_year_table(site_name, latest_month):
-    """任务2: 刷新 selection.{site}_flow_asin_365day，限定 latest_date_info=最新月 这批 asin 的9个每日变化字段"""
+    """任务2: 刷新 selection.{site}_flow_asin_365day，限定 latest_date_info=最新月 这批 asin 的每日变化字段"""
     table_name = f"{site_name}_flow_asin_365day"
     print(f"\n========== [任务2/年表] selection.{table_name} (latest_date_info={latest_month}) ==========")
     extra_where_sql = f"f.latest_date_info = '{latest_month}'"

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -230,6 +230,10 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `variation_bought_month`      INT              NULL     COMMENT '变体销量和(12个月变体销量之和口径见年度指标,此处为当月值)',
     `variation_bought_month_mom`  DECIMAL(20,4)    NULL     COMMENT '变体销量和环比',
     `variation_bought_month_yoy`  DECIMAL(20,4)    NULL     COMMENT '变体销量和同比',
+    `keepa_type`                  STRING           NULL     COMMENT 'Keepa type（小写、_转-）',
+    `keepa_item_type`             STRING           NULL     COMMENT 'Keepa itemTypeKeyword（小写、_转-）',
+    `keepa_type_cn`               STRING           NULL     COMMENT 'keepa_type 中文翻译',
+    `keepa_item_type_cn`          STRING           NULL     COMMENT 'keepa_item_type 中文翻译',
     INDEX idx_title      (`title`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引',
     INDEX idx_title_stem (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
     INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
@@ -1027,7 +1031,11 @@ SELECT
     addr.seller_city,
     f.variation_bought_month,
     f.variation_bought_month_mom,
-    f.variation_bought_month_yoy
+    f.variation_bought_month_yoy,
+    kp.keepa_type,
+    kp.keepa_item_type,
+    kp.keepa_type_cn,
+    kp.keepa_item_type_cn
 FROM `dwt`.`{site_name}_flow_asin_month` f
 LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
     ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
@@ -1248,6 +1256,10 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `variation_bought_month`            INT              NULL     COMMENT '变体销量和(当月值，非最新月置空)',
     `variation_bought_month_mom`        DECIMAL(20,4)    NULL     COMMENT '变体销量和环比(非最新月置空)',
     `variation_bought_month_yoy`        DECIMAL(20,4)    NULL     COMMENT '变体销量和同比(非最新月置空)',
+    `keepa_type`                        STRING           NULL     COMMENT 'Keepa type（小写、_转-）',
+    `keepa_item_type`                   STRING           NULL     COMMENT 'Keepa itemTypeKeyword（小写、_转-）',
+    `keepa_type_cn`                     STRING           NULL     COMMENT 'keepa_type 中文翻译',
+    `keepa_item_type_cn`                STRING           NULL     COMMENT 'keepa_item_type 中文翻译',
     INDEX idx_title      (`title`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引',
     INDEX idx_title_stem (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
     INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
@@ -1419,7 +1431,11 @@ SELECT
     -- asin本月快照没跟上最新月（latest_date_info != 本次跑批月份）时，变体销量和/环比/同比置空，跟asin_bought_month同理
     CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month ELSE NULL END AS variation_bought_month,
     CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month_mom ELSE NULL END AS variation_bought_month_mom,
-    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month_yoy ELSE NULL END AS variation_bought_month_yoy
+    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month_yoy ELSE NULL END AS variation_bought_month_yoy,
+    kp.keepa_type,
+    kp.keepa_item_type,
+    kp.keepa_type_cn,
+    kp.keepa_item_type_cn
 FROM `dwt`.`{site_name}_flow_asin_365day` f
 LEFT JOIN `dwt`.`{site_name}_flow_asin_365day_extra` agg ON f.asin = agg.asin
 -- ===== 利润率 =====