流量选品周计算-同环比计算增加基准月

Pyspark_job/dwt/dwt_flow_asin_week.py

@@ -51,9 +51,15 @@ class DwtFlowAsinWeek(Templates):
         self.spark = self.create_spark_object(
             app_name=f"{self.hive_tb}: {self.site_name}, {self.date_type}, {self.date_info}")
         # 前1/2/3周、4周前、去年同周、本周最后一天
-        self.last_week, self.last_2_week, self.last_3_week, self.previous_week, self.last_year_week, self.week_end_date = self.get_week_date_info()
+        self.last_week, self.last_2_week, self.last_3_week, self.previous_week, self.last_year_week, self.week_end_date, self.week_month = self.get_week_date_info()
         print(f"前1/2/3周: {self.last_week}/{self.last_2_week}/{self.last_3_week}, 4周前: {self.previous_week}, "
-              f"去年同周: {self.last_year_week}, 本周最后一天: {self.week_end_date}")
+              f"去年同周: {self.last_year_week}, 本周最后一天: {self.week_end_date}, 本周所属月: {self.week_month}")
+        # 基准周之前无数据，对比周早于基准周的指标不计算
+        self.base_week = "2026-41"
+        self.can_wow = self.last_week >= self.base_week
+        self.can_mom = self.previous_week >= self.base_week
+        self.can_yoy = self.last_year_week >= self.base_week
+        print(f"基准周: {self.base_week}, 是否计算 wow/mom/yoy: {self.can_wow}/{self.can_mom}/{self.can_yoy}")
         self.launch_time_interval_dict = self.get_launch_time_interval_dict(self.week_end_date)
         # 写入、分区初始化
         self.df_save = self.spark.sql(f"select 1+1;")
@@ -74,8 +80,8 @@ class DwtFlowAsinWeek(Templates):
         self.color_set = set()  # 颜色词表，read_data 阶段填充
 
     def get_week_date_info(self):
-        """获取前1/2/3周、4周前、去年同周、本周最后一天"""
-        df = self.spark.sql(f"select id, `date`, year_week, week_day from dim_date_20_to_30").toPandas()
+        """获取前1/2/3周、4周前、去年同周、本周最后一天、本周所属月"""
+        df = self.spark.sql(f"select id, `date`, year_week, week_day, year_month from dim_date_20_to_30").toPandas()
         current_week_id = int(list(df.loc[(df.year_week == self.date_info) & (df.week_day == 1)].id)[0])
         last_week = list(df.loc[df.id == current_week_id - 7].year_week)[0]
         last_2_week = list(df.loc[df.id == current_week_id - 14].year_week)[0]
@@ -84,7 +90,8 @@ class DwtFlowAsinWeek(Templates):
         week_end_date = str(list(df.loc[(df.year_week == self.date_info) & (df.week_day == 7)].date)[0])
         year, week = self.date_info.split("-")
         last_year_week = f"{int(year) - 1}-{week}"
-        return last_week, last_2_week, last_3_week, previous_week, last_year_week, week_end_date
+        week_month = list(df.loc[(df.year_week == self.date_info) & (df.week_day == 1)].year_month)[0]
+        return last_week, last_2_week, last_3_week, previous_week, last_year_week, week_end_date, week_month
 
     @staticmethod
     def get_launch_time_interval_dict(base_date):
@@ -142,9 +149,18 @@ class DwtFlowAsinWeek(Templates):
         print(f"{table_name} {date_type} 最新分区: {latest_date_info}")
         return latest_date_info
 
+    def get_month_date_info(self, table_name, date_type):
+        """优先取本周所属月分区，不存在时取最新分区"""
+        sql = f"show partitions {table_name} partition(site_name='{self.site_name}', date_type='{date_type}', date_info='{self.week_month}')"
+        if self.spark.sql(sql).count() > 0:
+            print(f"{table_name} {date_type} 使用本周所属月分区: {self.week_month}")
+            return self.week_month
+        print(f"{table_name} {date_type} 不存在 {self.week_month} 分区，改取最新分区")
+        return self.get_latest_date_info(table_name, date_type)
+
     def read_data(self):
-        print("1.获取dwd_asin_measure(最新month_week),得到各种类型的统计、ao值、自然流量占比、月销信息、bsr销量信息")
-        measure_date_info = self.get_latest_date_info("dwd_asin_measure", "month_week")
+        print("1.获取dwd_asin_measure(本周所属月month_week,无则取最新),得到各种类型的统计、ao值、自然流量占比、月销信息、bsr销量信息")
+        measure_date_info = self.get_month_date_info("dwd_asin_measure", "month_week")
         sql = f"""
             select asin, round(asin_ao_val, 3) as asin_ao_val, round(asin_ao_val_matrix, 3) as matrix_ao_val,
             asin_zr_counts, asin_sp_counts, (asin_sb1_counts + asin_sb2_counts) as asin_sb_counts,
@@ -195,59 +211,62 @@ class DwtFlowAsinWeek(Templates):
         self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
         self.df_fb_asin_info = self.df_fb_asin_info.repartition(100).persist(StorageLevel.DISK_ONLY)
         self.df_fb_asin_info.show(10, truncate=False)
-        print("5.获取上周整合结果(_rise、近7天环比_wow)")
-        sql = f"""
-            select asin, round(asin_ao_val, 3) as lastweek_asin_ao_val, asin_price as lastweek_asin_price,
-            sales as lastweek_sales, variation_num as lastweek_variation_num, asin_rating as lastweek_asin_rating,
-            bsr_orders as lastweek_bsr_orders, asin_total_comments as lastweek_asin_total_comments,
-            first_category_rank as lastweek_first_category_rank, asin_bought_month as lastweek_asin_bought_month,
-            variation_bought_month as lastweek_variation_bought_month, asin_bought_wow as lastweek_asin_bought_wow
-            from {self.hive_tb}
-            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_week}'
-            """
-        print("sql:" + sql)
-        self.df_flow_asin_last_week = self.spark.sql(sqlQuery=sql)
-        self.df_flow_asin_last_week = self.df_flow_asin_last_week.repartition(100).persist(StorageLevel.DISK_ONLY)
-        self.df_flow_asin_last_week.show(10, truncate=False)
-        print("5.1获取前2/3周月销_wow")
-        sql = f"""
-            select asin,
-            max(case when date_info = '{self.last_2_week}' then asin_bought_wow end) as asin_bought_wow_last_2w,
-            max(case when date_info = '{self.last_3_week}' then asin_bought_wow end) as asin_bought_wow_last_3w
-            from {self.hive_tb}
-            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info in ('{self.last_2_week}', '{self.last_3_week}')
-            group by asin
-            """
-        print("sql:" + sql)
-        self.df_bought_history = self.spark.sql(sqlQuery=sql)
-        self.df_bought_history = self.df_bought_history.repartition(100).persist(StorageLevel.DISK_ONLY)
-        self.df_bought_history.show(10, truncate=False)
-        print("5.2获取4周前整合结果(近30天环比_mom)")
-        sql = f"""
-            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,
-            variation_bought_month as previous_variation_bought_month from {self.hive_tb}
-            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.previous_week}'
-            """
-        print("sql:" + sql)
-        self.df_flow_asin_last = self.spark.sql(sqlQuery=sql)
-        self.df_flow_asin_last = self.df_flow_asin_last.repartition(100).persist(StorageLevel.DISK_ONLY)
-        self.df_flow_asin_last.show(10, truncate=False)
-        print("6.获取同比去年同周整合结果")
-        sql = f"""
-            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,
-            variation_bought_month as lastyear_variation_bought_month from {self.hive_tb}
-            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_year_week}'
-            """
-        print("sql:" + sql)
-        self.df_flow_asin_last_year = self.spark.sql(sqlQuery=sql)
-        self.df_flow_asin_last_year = self.df_flow_asin_last_year.repartition(100).persist(StorageLevel.DISK_ONLY)
-        self.df_flow_asin_last_year.show(10, truncate=False)
+        if self.can_wow:
+            print("5.获取上周整合结果(_rise、近7天环比_wow)")
+            sql = f"""
+                select asin, round(asin_ao_val, 3) as lastweek_asin_ao_val, asin_price as lastweek_asin_price,
+                sales as lastweek_sales, variation_num as lastweek_variation_num, asin_rating as lastweek_asin_rating,
+                bsr_orders as lastweek_bsr_orders, asin_total_comments as lastweek_asin_total_comments,
+                first_category_rank as lastweek_first_category_rank, asin_bought_month as lastweek_asin_bought_month,
+                variation_bought_month as lastweek_variation_bought_month, asin_bought_wow as lastweek_asin_bought_wow
+                from {self.hive_tb}
+                where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_week}'
+                """
+            print("sql:" + sql)
+            self.df_flow_asin_last_week = self.spark.sql(sqlQuery=sql)
+            self.df_flow_asin_last_week = self.df_flow_asin_last_week.repartition(100).persist(StorageLevel.DISK_ONLY)
+            self.df_flow_asin_last_week.show(10, truncate=False)
+            print("5.1获取前2/3周月销_wow")
+            sql = f"""
+                select asin,
+                max(case when date_info = '{self.last_2_week}' then asin_bought_wow end) as asin_bought_wow_last_2w,
+                max(case when date_info = '{self.last_3_week}' then asin_bought_wow end) as asin_bought_wow_last_3w
+                from {self.hive_tb}
+                where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info in ('{self.last_2_week}', '{self.last_3_week}')
+                group by asin
+                """
+            print("sql:" + sql)
+            self.df_bought_history = self.spark.sql(sqlQuery=sql)
+            self.df_bought_history = self.df_bought_history.repartition(100).persist(StorageLevel.DISK_ONLY)
+            self.df_bought_history.show(10, truncate=False)
+        if self.can_mom:
+            print("5.2获取4周前整合结果(近30天环比_mom)")
+            sql = f"""
+                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,
+                variation_bought_month as previous_variation_bought_month from {self.hive_tb}
+                where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.previous_week}'
+                """
+            print("sql:" + sql)
+            self.df_flow_asin_last = self.spark.sql(sqlQuery=sql)
+            self.df_flow_asin_last = self.df_flow_asin_last.repartition(100).persist(StorageLevel.DISK_ONLY)
+            self.df_flow_asin_last.show(10, truncate=False)
+        if self.can_yoy:
+            print("6.获取同比去年同周整合结果")
+            sql = f"""
+                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,
+                variation_bought_month as lastyear_variation_bought_month from {self.hive_tb}
+                where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_year_week}'
+                """
+            print("sql:" + sql)
+            self.df_flow_asin_last_year = self.spark.sql(sqlQuery=sql)
+            self.df_flow_asin_last_year = self.df_flow_asin_last_year.repartition(100).persist(StorageLevel.DISK_ONLY)
+            self.df_flow_asin_last_year.show(10, truncate=False)
         print("7.读取最新keepa数据")
         sql = f"""
         select asin, tracking_since, keepa_launch_time from dim_keepa_asin_info where site_name = '{self.site_name}'
@@ -585,9 +604,13 @@ class DwtFlowAsinWeek(Templates):
 
     # 处理所有变化率相关指标（_rise/_wow对比上周、_mom对比4周前、_yoy对比去年同周）
     def handle_change_rate(self):
-        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_week, on=['asin'], how='left')
-        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last, on=['asin'], how='left')
-        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_year, on=['asin'], how='left')
+        if self.can_wow:
+            self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_week, on=['asin'], how='left')
+        if self.can_mom:
+            self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last, on=['asin'], how='left')
+        if self.can_yoy:
+            self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_year, on=['asin'], how='left')
+        null_double = F.lit(None).cast(DoubleType())
         # (current_col, base_col, suffix, rise_round)
         # base_col 为不带前缀的历史字段名；rise_round: None 表示不计算 _rise；整数表示保留小数位；'int' 表示转 IntegerType
         columns_to_change = [
@@ -603,25 +626,39 @@ class DwtFlowAsinWeek(Templates):
             ("variation_bought_month", "variation_bought_month", "variation_bought_month", "int"),
         ]
         for current_col, base_col, suffix, rise_round in columns_to_change:
-            rise_col, wow_col = self.calculate_change(current_col, f"lastweek_{base_col}")
-            _, mom_col = self.calculate_change(current_col, f"previous_{base_col}")
-            _, yoy_col = self.calculate_change(current_col, f"lastyear_{base_col}")
-            if rise_round is not None:
+            if self.can_wow:
+                rise_col, wow_col = self.calculate_change(current_col, f"lastweek_{base_col}")
                 if rise_round == "int":
-                    self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))
-                else:
-                    self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", F.round(rise_col, rise_round))
+                    rise_col = rise_col.cast(IntegerType())
+                elif rise_round is not None:
+                    rise_col = F.round(rise_col, rise_round)
+                wow_col = F.round(wow_col, 4)
+            else:
+                rise_col = F.lit(None).cast(IntegerType()) if rise_round == "int" else null_double
+                wow_col = null_double
+            if rise_round is not None:
+                self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col)
+            mom_col = F.round(self.calculate_change(current_col, f"previous_{base_col}")[1], 4) if self.can_mom else null_double
+            yoy_col = F.round(self.calculate_change(current_col, f"lastyear_{base_col}")[1], 4) if self.can_yoy else null_double
             self.df_asin_detail = self.df_asin_detail \
-                .withColumn(f"{suffix}_wow", F.round(wow_col, 4)) \
-                .withColumn(f"{suffix}_mom", F.round(mom_col, 4)) \
-                .withColumn(f"{suffix}_yoy", F.round(yoy_col, 4))
+                .withColumn(f"{suffix}_wow", wow_col) \
+                .withColumn(f"{suffix}_mom", mom_col) \
+                .withColumn(f"{suffix}_yoy", yoy_col)
         # 前1/2/3周月销_wow，供后端判断连续上升
-        self.df_asin_detail = self.df_asin_detail.join(self.df_bought_history, on=['asin'], how='left') \
-            .withColumn("asin_bought_wow_last_1w", F.col("lastweek_asin_bought_wow"))
-        self.df_flow_asin_last_week.unpersist()
-        self.df_flow_asin_last.unpersist()
-        self.df_flow_asin_last_year.unpersist()
-        self.df_bought_history.unpersist()
+        if self.can_wow:
+            self.df_asin_detail = self.df_asin_detail.join(self.df_bought_history, on=['asin'], how='left') \
+                .withColumn("asin_bought_wow_last_1w", F.col("lastweek_asin_bought_wow"))
+            self.df_flow_asin_last_week.unpersist()
+            self.df_bought_history.unpersist()
+        else:
+            self.df_asin_detail = self.df_asin_detail \
+                .withColumn("asin_bought_wow_last_1w", null_double) \
+                .withColumn("asin_bought_wow_last_2w", null_double) \
+                .withColumn("asin_bought_wow_last_3w", null_double)
+        if self.can_mom:
+            self.df_flow_asin_last.unpersist()
+        if self.can_yoy:
+            self.df_flow_asin_last_year.unpersist()
 
     def handle_asin_different_source(self):
         self.df_asin_detail = self.df_asin_detail.join(