流量选品月-增加竞卖指标相关计算

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -308,7 +308,8 @@ def read_and_normalize_month_data(spark, site_name, date_info):
             asin_describe, asin_fbm_price, describe_len,
             title_matching_degree,
             multi_color_flag, multi_color_str,
-            amazon_label
+            amazon_label,
+            auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all
         FROM dwt_flow_asin
         WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
     """
@@ -447,6 +448,11 @@ def read_and_normalize_month_data(spark, site_name, date_info):
         F.col('multi_color_str'),
         # ===== Amazon标签 =====
         F.col('amazon_label'),
+        # ===== 竞卖与SKU创建 =====
+        F.col('auctions_num').cast('int').alias('auctions_num'),
+        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'),
     ).cache()
 
     count = df_save.count()
@@ -478,7 +484,8 @@ def write_dwt_month_table(df_save, doris_table):
         "asin_bought_month, asin_lqs_rating, asin_lqs_rating_detail, asin_lob_info, is_contains_lob_info, "
         "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"
+        "title_matching_degree, multi_color_flag, multi_color_str, amazon_label, "
+        "auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all"
     )
     print(f"[Step 3] 写入 Doris {DORIS_DB}.{doris_table}")
     DorisHelper.spark_export_with_columns(
@@ -612,7 +619,11 @@ CREATE TABLE IF NOT EXISTS `dwt`.`{site_name}_flow_asin_365day`
     `title_matching_degree`       DECIMAL(20,4),
     `multi_color_flag`            TINYINT          DEFAULT "0",
     `multi_color_str`             STRING,
-    `amazon_label`                STRING
+    `amazon_label`                STRING,
+    `auctions_num`                INT,
+    `auctions_num_all`            INT,
+    `skus_num_creat`              INT,
+    `skus_num_creat_all`          INT
 )
 ENGINE=OLAP
 UNIQUE KEY(`asin`)
@@ -651,7 +662,8 @@ STG_INSERT_COLUMNS = """
     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
+    multi_color_flag, multi_color_str, amazon_label,
+    auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all
 """.strip()
 
 
@@ -942,10 +954,10 @@ SELECT
     f.asin_type,
     uma.usr_mask_progress,
     COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
-    COALESCE(aa.auctions_num,       0) AS auctions_num,
-    COALESCE(aa.auctions_num_all,   0) AS auctions_num_all,
-    COALESCE(aa.skus_num_creat,     0) AS skus_num_creat,
-    COALESCE(aa.skus_num_creat_all, 0) AS skus_num_creat_all,
+    COALESCE(f.auctions_num,       0) AS auctions_num,
+    COALESCE(f.auctions_num_all,   0) AS auctions_num_all,
+    COALESCE(f.skus_num_creat,     0) AS skus_num_creat,
+    COALESCE(f.skus_num_creat_all, 0) AS skus_num_creat_all,
     f.title_matching_degree,
     bb.brand_badge_reason,
     ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
@@ -963,7 +975,6 @@ LEFT JOIN (
     FROM `dwd`.`dwd_asin_source_flag`
     WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
 ) cf ON f.asin = cf.asin
-LEFT JOIN `dwd`.`dwd_asin_auction` aa ON f.asin = aa.asin
 LEFT JOIN `dwd`.`dwd_st_brand_badge` bb
     ON f.brand = bb.brand AND bb.site_name = '{site_name}' AND bb.date_info = '{date_info}'
 -- ===== 卖家地址（按 site_name 查对应分区） =====
@@ -1284,10 +1295,10 @@ SELECT
     f.asin_type,
     uma.usr_mask_progress,
     COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
-    COALESCE(aa.auctions_num,       0) AS auctions_num,
-    COALESCE(aa.auctions_num_all,   0) AS auctions_num_all,
-    COALESCE(aa.skus_num_creat,     0) AS skus_num_creat,
-    COALESCE(aa.skus_num_creat_all, 0) AS skus_num_creat_all,
+    COALESCE(f.auctions_num,       0) AS auctions_num,
+    COALESCE(f.auctions_num_all,   0) AS auctions_num_all,
+    COALESCE(f.skus_num_creat,     0) AS skus_num_creat,
+    COALESCE(f.skus_num_creat_all, 0) AS skus_num_creat_all,
     f.title_matching_degree,
     bb.brand_badge_reason,
     ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
@@ -1318,8 +1329,6 @@ LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
 -- ===== 用户标记 ASIN/品类 =====
 LEFT JOIN `selection`.`user_mask_asin`     uma ON f.asin        = uma.asin
 LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
--- ===== 拍卖/SKU =====
-LEFT JOIN `dwd`.`dwd_asin_auction` aa ON f.asin = aa.asin
 -- ===== 品牌推荐原因——
 LEFT JOIN (
     SELECT brand, brand_badge_reason

Pyspark_job/dwt/dwt_flow_asin.py

@@ -8,6 +8,7 @@
                   dim_asin_bs_category
                   dim_fb_asin_info
                   dim_asin_volume
+                  product_audit_asin_sku(mysql)
 
 
    @SinkTable   : dwt_flow_asin
@@ -312,6 +313,22 @@ class DwtFlowAsin(Templates):
         self.color_set = {row.en_name for row in color_rows}
         print(f"颜色词表共 {len(self.color_set)} 条")
 
+        print("12.获取product_audit_asin_sku,得到asin竞卖数、sku创建数")
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        sql = """
+            select asin,
+            count(asin) as auctions_num,
+            count((case when sku != '' then sku end)) as skus_num_creat
+            from product_audit_asin_sku
+            WHERE asin REGEXP '^[0-9A-Z]{10}$'
+            group by asin
+        """
+        print("sql:" + sql)
+        self.df_asin_auction = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+                                                          username=mysql_con['username'], query=sql)
+        self.df_asin_auction = self.df_asin_auction.repartition(60).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_auction.show(10, truncate=False)
+
     # 处理asin基础属性信息(体积重量相关)及bsr销售额相关信息
     def handle_asin_basic_attribute(self):
         # 处理重量类型
@@ -449,6 +466,33 @@ class DwtFlowAsin(Templates):
             F.coalesce(F.col("variation_bought_month"), F.col("asin_bought_month").cast("int"))
         )
 
+    # 处理竞卖数据(auctions_num/skus_num_creat)及母体竞卖数据(auctions_num_all/skus_num_creat_all)
+    def handle_asin_auction(self):
+        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_auction, on=['asin'], how='left')
+        self.df_asin_auction.unpersist()
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "auctions_num", F.col("auctions_num").cast("int")
+        ).withColumn(
+            "skus_num_creat", F.col("skus_num_creat").cast("int")
+        ).na.fill({"auctions_num": 0, "skus_num_creat": 0})
+        # 母体竞卖数：按parent_asin聚合子体auctions_num/skus_num_creat；asin等于parent_asin且variation_num>0的不参与求和
+        df_auction_all = self.df_asin_detail.filter(
+            "parent_asin is not null"
+        ).filter(
+            ~((F.col("asin") == F.col("parent_asin")) & (F.col("variation_num") > 0))
+        ).select("parent_asin", "auctions_num", "skus_num_creat")
+        df_auction_all_agg = df_auction_all.groupby(['parent_asin']).agg(
+            F.sum("auctions_num").cast("int").alias("auctions_num_all"),
+            F.sum("skus_num_creat").cast("int").alias("skus_num_creat_all")
+        )
+        self.df_asin_detail = self.df_asin_detail.join(
+            df_auction_all_agg, on=['parent_asin'], how='left'
+        ).withColumn(
+            "auctions_num_all", F.coalesce(F.col("auctions_num_all"), F.col("auctions_num"))
+        ).withColumn(
+            "skus_num_creat_all", F.coalesce(F.col("skus_num_creat_all"), F.col("skus_num_creat"))
+        )
+
     def handle_parent_asin_variation(self):
         """处理父ASIN变体聚合数据，结果存入 self.df_parent_asin_variat_agg"""
         if self.date_type not in ['month', 'month_week'] or self.date_info < '2024-06':
@@ -463,8 +507,10 @@ class DwtFlowAsin(Templates):
             .withColumn("p_rank", F.row_number().over(latest_window))
             .filter(F.col("p_rank") == 1).drop("p_rank")
             .select("parent_asin", "asin_crawl_date", "matrix_ao_val", "matrix_flow_proportion",
+                    "auctions_num_all", "skus_num_creat_all",
                     F.explode("variat_list_parsed").alias("variant_attribute"))
             .select("parent_asin", "asin_crawl_date", "matrix_ao_val", "matrix_flow_proportion",
+                    "auctions_num_all", "skus_num_creat_all",
                     F.col("variant_attribute")[0].alias("asin"),
                     F.col("variant_attribute")[1].alias("color"),
                     F.col("variant_attribute")[3].alias("size"),
@@ -474,6 +520,8 @@ class DwtFlowAsin(Templates):
             F.first("asin_crawl_date").alias("asin_crawl_date"),
             F.first("matrix_ao_val").alias("matrix_ao_val"),
             F.first("matrix_flow_proportion").alias("matrix_flow_proportion"),
+            F.first("auctions_num_all").alias("auctions_num_all"),
+            F.first("skus_num_creat_all").alias("skus_num_creat_all"),
             F.concat_ws(',', F.collect_list("asin")).alias("variation_info"),
             F.to_json(F.collect_list(F.struct(F.col("color"), F.col("size"), F.col("style")))).alias("attr_info")
         )
@@ -484,7 +532,8 @@ class DwtFlowAsin(Templates):
             "variation_info", "attr_info",
             F.current_timestamp().alias("updated_at"),
             F.round(F.col("matrix_flow_proportion"), 4).alias("matrix_flow_proportion"),
-            F.round(F.col("matrix_ao_val"), 4).alias("matrix_ao_val")
+            F.round(F.col("matrix_ao_val"), 4).alias("matrix_ao_val"),
+            "auctions_num_all", "skus_num_creat_all"
         ).persist(StorageLevel.DISK_ONLY)
 
     # 处理配送方式、卖家所在地以及卖家所在地类型
@@ -827,6 +876,7 @@ class DwtFlowAsin(Templates):
                    "asin_source_flag", "bsr_last_seen_at", "bsr_seen_count_30d", "nsr_last_seen_at", "nsr_seen_count_30d",
                    "multi_color_flag", "multi_color_str", "amazon_label", "asin_weight_str", "best_sellers_herf", "best_sellers_rank",
                    "variation_bought_month", "variation_bought_month_mom", "variation_bought_month_yoy",
+                   "auctions_num", "auctions_num_all", "skus_num_creat", "skus_num_creat_all",
                    F.lit(self.site_name).alias("site_name"), F.lit(self.date_type).alias("date_type"),
                    F.lit(self.date_info).alias("date_info"))
         self.df_save = self.df_save.na.fill(
@@ -889,7 +939,8 @@ class DwtFlowAsin(Templates):
                            "asin_describe", "asin_fbm_price",
                            F.col("describe_len").alias("asin_describe_len"),
                            "asin_type",
-                           F.lit(None).cast("int").alias("is_amazon_new")
+                           F.lit(None).cast("int").alias("is_amazon_new"),
+                           "auctions_num", "auctions_num_all", "skus_num_creat", "skus_num_creat_all"
                            )
                 table_columns = """asin, asin_ao_val, asin_title, asin_title_len, asin_category_desc, asin_volume,
                           asin_weight, asin_weight_str, asin_launch_time, asin_brand_name,
@@ -904,11 +955,12 @@ class DwtFlowAsin(Templates):
                           bsr_orders, bsr_orders_sale, page_inventory, asin_bought_month,
                           seller_json, buy_box_seller_type,
                           asin_describe, asin_fbm_price, asin_describe_len,
-                          asin_type, is_amazon_new"""
+                          asin_type, is_amazon_new,
+                          auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all"""
                 DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.asin_latest_detail_table, table_columns=table_columns)
                 print("save asin_latest_detail success")
                 print("往doris存储父ASIN最新变体信息：")
-                table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info, updated_at, matrix_flow_proportion, matrix_ao_val"
+                table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info, updated_at, matrix_flow_proportion, matrix_ao_val, auctions_num_all, skus_num_creat_all"
                 DorisHelper.spark_export_with_columns(
                     df_save=self.df_parent_asin_variat_agg, db_name=self.doris_db,
                     table_name=self.parent_asin_latest_detail_table, table_columns=table_columns
@@ -939,6 +991,7 @@ class DwtFlowAsin(Templates):
         self.handle_asin_detail_all_type()
         self.handle_asin_category_info()
         self.handle_asin_measure()
+        self.handle_asin_auction()
         # 打断 lazy 链：measure 阶段含自连接，落盘后释放，避免后续 action 触发时计划树过长导致 OOM
         df_measure_done = self.df_asin_detail
         self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)