流量选品-新增变体销量和及同环比

Pyspark_job/dwt/dwt_flow_asin.py

@@ -256,7 +256,8 @@ class DwtFlowAsin(Templates):
             select asin, round(asin_ao_val, 3) as previous_asin_ao_val, asin_price as previous_asin_price,
             sales as previous_sales, variation_num as previous_variation_num, asin_rating as previous_asin_rating,
             bsr_orders as previous_bsr_orders, asin_total_comments as previous_asin_total_comments,
-            first_category_rank as previous_first_category_rank, asin_bought_month as previous_asin_bought_month from dwt_flow_asin
+            first_category_rank as previous_first_category_rank, asin_bought_month as previous_asin_bought_month,
+            variation_bought_month as previous_variation_bought_month from dwt_flow_asin
             where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.previous_date}'
             """
         print("sql:" + sql)
@@ -268,7 +269,8 @@ class DwtFlowAsin(Templates):
             select asin, round(asin_ao_val, 3) as lastyear_asin_ao_val, asin_price as lastyear_asin_price,
             sales as lastyear_sales, variation_num as lastyear_variation_num, asin_rating as lastyear_asin_rating,
             bsr_orders as lastyear_bsr_orders, asin_total_comments as lastyear_asin_total_comments,
-            first_category_rank as lastyear_first_category_rank, asin_bought_month as lastyear_asin_bought_month from dwt_flow_asin
+            first_category_rank as lastyear_first_category_rank, asin_bought_month as lastyear_asin_bought_month,
+            variation_bought_month as lastyear_variation_bought_month from dwt_flow_asin
             where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info_last_year}'
             """
         print("sql:" + sql)
@@ -430,6 +432,23 @@ class DwtFlowAsin(Templates):
         self.df_asin_measure.unpersist()
         df_ao_stage.unpersist()
 
+        # 变体销量和(variation_bought_month)：按parent_asin聚合asin_bought_month；
+        # asin等于parent_asin且variation_num>0的不参与求和
+        df_variation_bought_month = 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", "asin_bought_month")
+        df_variation_bought_month_agg = df_variation_bought_month.groupby(['parent_asin']).agg(
+            F.sum("asin_bought_month").cast("int").alias("variation_bought_month")
+        )
+        self.df_asin_detail = self.df_asin_detail.join(
+            df_variation_bought_month_agg, on=['parent_asin'], how='left'
+        ).withColumn(
+            "variation_bought_month",
+            F.coalesce(F.col("variation_bought_month"), F.col("asin_bought_month").cast("int"))
+        )
+
     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':
@@ -622,6 +641,7 @@ class DwtFlowAsin(Templates):
             ("asin_price", "previous_asin_price", "lastyear_asin_price", "asin_price", 2),
             ("sales", "previous_sales", "lastyear_sales", "asin_sales", 2),
             ("asin_bought_month", "previous_asin_bought_month", "lastyear_asin_bought_month", "asin_bought", None),
+            ("variation_bought_month", "previous_variation_bought_month", "lastyear_variation_bought_month", "variation_bought_month", None),
         ]
         for current_col, prev_col, lastyear_col, suffix, rise_round in columns_to_change:
             rise_col, mom_col = self.calculate_change(current_col, prev_col)
@@ -634,7 +654,8 @@ class DwtFlowAsin(Templates):
             self.df_asin_detail = self.df_asin_detail \
                 .withColumn(f"{suffix}_mom", F.round(mom_col, 4)) \
                 .withColumn(f"{suffix}_yoy", F.round(yoy_col, 4))
-        self.df_asin_detail = self.df_asin_detail.drop('previous_asin_bought_month', 'lastyear_asin_bought_month')
+        self.df_asin_detail = self.df_asin_detail.drop('previous_asin_bought_month', 'lastyear_asin_bought_month',
+                                                        'previous_variation_bought_month', 'lastyear_variation_bought_month')
         self.df_flow_asin_last.unpersist()
         self.df_flow_asin_last_year.unpersist()
 
@@ -805,6 +826,7 @@ class DwtFlowAsin(Templates):
                    "asin_bought_mom", "asin_bought_yoy", "describe_len", "tracking_since", "tracking_since_type",
                    "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",
                    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(