流量选品-新增字段 变体销量和

Pyspark_job/dwd/dwd_st_asin_measure.py

@@ -178,7 +178,7 @@ class DwdStMeasure(Templates):
         self.df_asin_bs = self.spark.sql(sql).cache()
         self.df_asin_bs.show(10)
 
-        sql = f"select asin, asin_title, asin_price, parent_asin, asin_bought_month, updated_time, current_asin " \
+        sql = f"select asin, asin_title, asin_price, parent_asin, asin_bought_month, updated_time, current_asin, variation_num " \
               f"from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type.replace('_old', '')}' and date_info='{self.date_info}';"
         print("sql:", sql)
         self.df_asin_detail = self.spark.sql(sql).cache()
@@ -659,7 +659,7 @@ class DwdStMeasure(Templates):
             F.when(F.col("asin_st_counts").isNotNull(), F.round(F.col("asin_zr_counts") / F.col("asin_st_counts"), 3))
         )
         self.df_save_asin = self.df_save_asin.join(
-            self.df_asin_detail.select("asin", "parent_asin"), on='asin', how='left'
+            self.df_asin_detail.select("asin", "parent_asin", "variation_num"), on='asin', how='left'
         )
         # 4.计算asin的母体ao值和母体zr流量占比
         df_asin_variation = self.df_save_asin.filter("parent_asin is not null").select("parent_asin",
@@ -686,16 +686,32 @@ class DwdStMeasure(Templates):
         self.df_save_asin = self.df_save_asin.join(
             df_asin_variation_agg, on=['parent_asin'], how='left'
         )
-        # 5.若母体自然流量占比为null，则用asin的自然流量占比替代，ao同理
+        # 4.5 计算变体销量和(variation_bouhgt_month)：
+        # 如果asin等于parent_asin，且variation_num>0，不参与sum统计
+        df_variation_bought_month = self.df_save_asin.filter(
+            "parent_asin is not null"
+        ).filter(
+            ~((F.col("asin") == F.col("parent_asin")) & (F.col("variation_num") > 0))
+        ).select("parent_asin", "asin_amazon_orders")
+        df_variation_bought_month_agg = df_variation_bought_month.groupby(['parent_asin']).agg(
+            F.sum("asin_amazon_orders").cast("int").alias("variation_bouhgt_month")
+        )
+        self.df_save_asin = self.df_save_asin.join(
+            df_variation_bought_month_agg, on=['parent_asin'], how='left'
+        )
+        # 5.若母体自然流量占比为null，则用asin的自然流量占比替代，ao、variation_bouhgt_month同理
         self.df_save_asin = self.df_save_asin.withColumn(
             "asin_flow_proportion_matrix",
             F.coalesce(F.col("asin_flow_proportion_matrix"), F.col("asin_zr_flow_proportion"))
         ).withColumn(
             "asin_ao_val_matrix",
             F.coalesce(F.col("asin_ao_val_matrix"), F.col("asin_ao_val"))
+        ).withColumn(
+            "variation_bouhgt_month",
+            F.coalesce(F.col("variation_bouhgt_month"), F.col("asin_amazon_orders").cast("int"))
         )
         self.df_save_asin.show(10, truncate=False)
-        self.df_save_asin = self.df_save_asin.drop("parent_asin")
+        self.df_save_asin = self.df_save_asin.drop("parent_asin", "variation_num")
 
     def handle_st_ao_and_zr_flow_proportion(self):
         print("计算st维度的ao+zr流量占比")