流量选品月/年-增加变体销量

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -226,6 +226,9 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `title_stem_15`               STRING           NULL     COMMENT '标题词干前15个词(按空格分词截取)',
     `seller_province`             STRING           NULL     COMMENT '卖家所在省份(dim_seller_address关联,仅国内卖家有值)',
     `seller_city`                 STRING           NULL     COMMENT '卖家所在城市(dim_seller_address关联,仅国内卖家有值)',
+    `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 '变体销量和同比',
     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个词倒排索引',
@@ -297,6 +300,7 @@ def read_and_normalize_month_data(spark, site_name, date_info):
             seller_country_name,
             asin_type,
             asin_bought_month,
+            variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy,
             asin_lqs_rating, asin_lqs_rating_detail,
             asin_lob_info, is_contains_lob_info,
             is_package_quantity_abnormal,
@@ -453,6 +457,10 @@ def read_and_normalize_month_data(spark, site_name, date_info):
         F.col('auctions_num_all').cast('int').alias('auctions_num_all'),
         F.col('skus_num_creat').cast('int').alias('skus_num_creat'),
         F.col('skus_num_creat_all').cast('int').alias('skus_num_creat_all'),
+        # ===== 变体销量和及环比/同比（新增字段放末尾）=====
+        F.col('variation_bought_month').cast('int').alias('variation_bought_month'),
+        F.round(F.col('variation_bought_month_mom').cast('double'), 4).cast('decimal(20,4)').alias('variation_bought_month_mom'),
+        F.round(F.col('variation_bought_month_yoy').cast('double'), 4).cast('decimal(20,4)').alias('variation_bought_month_yoy'),
     ).cache()
 
     count = df_save.count()
@@ -485,7 +493,8 @@ def write_dwt_month_table(df_save, doris_table):
         "is_package_quantity_abnormal, zr_flow_proportion, matrix_flow_proportion, matrix_ao_val, "
         "product_features, img_info, collapse_asin, follow_sellers_count, asin_describe, fbm_price, describe_len, "
         "title_matching_degree, multi_color_flag, multi_color_str, amazon_label, "
-        "auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all"
+        "auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all, "
+        "variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy"
     )
     print(f"[Step 3] 写入 Doris {DORIS_DB}.{doris_table}")
     DorisHelper.spark_export_with_columns(
@@ -623,7 +632,10 @@ CREATE TABLE IF NOT EXISTS `dwt`.`{site_name}_flow_asin_365day`
     `auctions_num`                INT,
     `auctions_num_all`            INT,
     `skus_num_creat`              INT,
-    `skus_num_creat_all`          INT
+    `skus_num_creat_all`          INT,
+    `variation_bought_month`      INT,
+    `variation_bought_month_mom`  DECIMAL(20,4),
+    `variation_bought_month_yoy`  DECIMAL(20,4)
 )
 ENGINE=OLAP
 UNIQUE KEY(`asin`)
@@ -663,7 +675,8 @@ STG_INSERT_COLUMNS = """
     product_features, img_info, collapse_asin, follow_sellers_count,
     asin_describe, fbm_price, describe_len, title_matching_degree,
     multi_color_flag, multi_color_str, amazon_label,
-    auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all
+    auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all,
+    variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy
 """.strip()
 
 
@@ -1001,7 +1014,10 @@ SELECT
     bb.brand_badge_reason,
     ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
     addr.seller_province,
-    addr.seller_city
+    addr.seller_city,
+    f.variation_bought_month,
+    f.variation_bought_month_mom,
+    f.variation_bought_month_yoy
 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}'
@@ -1219,6 +1235,9 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `variation_bought_month_q4`         INT              NULL,
     `variation_bought_month_peak`       INT              NULL,
     `variation_peak_month_arr`          ARRAY<INT>       NULL,
+    `variation_bought_month`            INT              NULL     COMMENT '变体销量和(当月值，非最新月置空)',
+    `variation_bought_month_mom`        DECIMAL(20,4)    NULL     COMMENT '变体销量和环比(非最新月置空)',
+    `variation_bought_month_yoy`        DECIMAL(20,4)    NULL     COMMENT '变体销量和同比(非最新月置空)',
     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个词倒排索引',
@@ -1386,7 +1405,11 @@ SELECT
     agg.variation_bought_month_9, agg.variation_bought_month_10, agg.variation_bought_month_11, agg.variation_bought_month_12,
     agg.variation_bought_month_q1, agg.variation_bought_month_q2, agg.variation_bought_month_q3, agg.variation_bought_month_q4,
     agg.variation_bought_month_peak,
-    agg.variation_peak_month_arr
+    agg.variation_peak_month_arr,
+    -- 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
 FROM `dwt`.`{site_name}_flow_asin_365day` f
 LEFT JOIN `dwt`.`{site_name}_flow_asin_365day_extra` agg ON f.asin = agg.asin
 -- ===== 利润率 =====