新增创历史新高字段

Pyspark_job/dwt/dwt_aba_last_change_rate.py

@@ -17,6 +17,7 @@ from utils.hdfs_utils import HdfsUtils
 from utils.common_util import CommonUtil, DateTypes
 from utils.spark_util import SparkUtil
 from pyspark.sql import functions as F
+from pyspark.sql.window import Window
 
 
 class DwtAbaLastChangeRate(object):
@@ -51,6 +52,14 @@ class DwtAbaLastChangeRate(object):
         # 爬虫漏抓补 rank：dim_st_detail 上月/去年同月 rank，给环比/同比 last_rank/last_year_rank 兜底（仅 month）
         self.df_last_month_rank_dim = self.spark.sql(f"select 1+1;")  # 上月 rank(dim)
         self.df_last_year_rank_dim = self.spark.sql(f"select 1+1;")   # 去年同月 rank(dim)
+        # 创历史新高 6 个 flag（当月及往回 5 个月，每月是否为 2023-01 以来历史新高）
+        # 只在 us + month + date_info>=2025-07 计算真值；uk/de、非 month、早于 2025-07 全部占位 -1
+        self.is_new_high_calc = (
+            self.site_name == 'us'
+            and self.date_type == DateTypes.month.name
+            and self.date_info >= '2025-07'
+        )
+        self.df_new_high_hist = self.spark.sql(f"select 1+1;")  # 2023-01~当月全历史 rank(dim_st_detail) 占位
 
     def handle_date_offset(self, handle_type: int):
         # handle_type = 0 代表计算环比日期，等于 1 代表计算同比日期
@@ -88,12 +97,18 @@ class DwtAbaLastChangeRate(object):
             self.handle_base()
             self.handle_year_ratio()
 
-        # 需求1：仅 month 类型计算近6月排名变化（12 字段），其他类型补 null 占位（确保 save_data 不报错）
+        # 仅 month 类型计算近6月排名变化（12 字段），其他类型补 null 占位（确保 save_data 不报错）
         if self.date_type == DateTypes.month.name:
             self.handle_rank_rate_history()
         else:
             self.handle_rank_rate_padding()
 
+        # 创历史新高 6 flag：仅 us+month+>=2025-07 算真值，其余（uk/de、非 month、早于 25-07）占位 -1
+        if self.is_new_high_calc:
+            self.handle_new_high_history()
+        else:
+            self.handle_new_high_padding()
+
         self.save_data()
 
     def read_data(self):
@@ -165,7 +180,7 @@ class DwtAbaLastChangeRate(object):
         self.df_st_last_year_data = self.spark.sql(sql).repartition(40, 'search_term').cache()
         self.df_st_last_year_data.show(10, truncate=True)
 
-        # 需求1：仅 month 类型读取历史数据用于近6月排名变化计算
+        # 仅 month 类型读取历史数据用于近6月排名变化计算
         if self.date_type == DateTypes.month.name:
             # 计算 M-1 ~ M-6 月份字符串（如 2026-04 → ['2026-03', '2026-02', '2026-01', '2025-12', '2025-11', '2025-10']）
             month_list = [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 7)]
@@ -228,6 +243,26 @@ class DwtAbaLastChangeRate(object):
             """
             self.df_last_year_rank_dim = self.spark.sql(sql_last_year_rank).repartition(40, 'search_term').cache()
 
+        # 创历史新高：读该词 2023-01 ~ 当月的全历史月度 rank（仅 us+month+>=2025-07 才读）
+        # 是否历史新高：拿当月 rank 和该词 2023-01 以来所有月比，不是滚动窗口，取全历史
+        # 用 dim_st_detail：ABA 清洗表、不过爬虫过滤，漏抓月仍有该词 st_rank，避免假断连
+        # st_rank>0：滤脏数据/未在榜；无该月行 = 缺月（下游按"没创新高=0"处理）
+        if self.is_new_high_calc:
+            sql_new_high = f"""
+                select
+                    search_term,
+                    cast(st_rank as int) as rank,
+                    date_info
+                from dim_st_detail
+                where site_name = '{self.site_name}'
+                  and date_type = '{self.date_type}'
+                  and date_info >= '2023-01'
+                  and date_info <= '{self.date_info}'
+                  and st_rank > 0
+            """
+            # 只被 handle_new_high_history 的窗口消费一次：不 cache（84M 行缓存无收益且占内存），
+            # 也不预 repartition（窗口 partitionBy 会自带 shuffle，预分区反而多一次 84M shuffle）
+            self.df_new_high_hist = self.spark.sql(sql_new_high)
 
     def handle_base(self):
         self.df_st_base_data = self.df_aba_analytics.join(
@@ -322,7 +357,7 @@ class DwtAbaLastChangeRate(object):
         })
         self.df_save = df_year_ratio
 
-    # 需求1：近6月排名变化率/变化量（仅 month 类型调用）
+    # 近6月排名变化率/变化量（仅 month 类型调用）
     def handle_rank_rate_history(self):
         """
         基于 self.df_save（已含同比/环比所有字段）+ M-1~M-6 历史 rank，
@@ -390,7 +425,7 @@ class DwtAbaLastChangeRate(object):
         self.df_save = df
         self.df_hist_rank.unpersist()
 
-    # 需求1：非 month 类型（day/week/last30day/last365day）补 12 字段为 null
+    # 非 month 类型（day/week/last30day/last365day）补 12 字段为 null
     # 原因：Hive 表加了这 12 列后，所有分区 schema 都得有这些字段，否则 save_data 的 select 报错
     def handle_rank_rate_padding(self):
         new_cols = [
@@ -402,6 +437,52 @@ class DwtAbaLastChangeRate(object):
         for col_name in new_cols:
             self.df_save = self.df_save.withColumn(col_name, F.lit(None))
 
+    # 6 个字段名：当月 M、往回 M-1 ~ M-5
+    NEW_HIGH_COLS = [
+        'new_high_this_month',   # M
+        'new_high_1_month_ago',  # M-1
+        'new_high_2_month_ago',  # M-2
+        'new_high_3_month_ago',  # M-3
+        'new_high_4_month_ago',  # M-4
+        'new_high_5_month_ago',  # M-5
+    ]
+
+    # 创历史新高：当月及往回 5 个月，每月排名是否为 2023-01 以来历史新高（仅 us+month+>=2025-07 调用）
+    # 取值：1=创新高（追平/首现也算），0=没创新高（含下榜/缺月）
+    def handle_new_high_history(self):
+        # 按 search_term 开窗、date_info 升序，算当月之前所有在榜月的最小 rank（不含当月）
+        # 当月 rank <= 该最小值即创新高；首现月之前无历史，hist_min_before 为 null，记 1
+        w = Window.partitionBy('search_term').orderBy('date_info').rowsBetween(Window.unboundedPreceding, -1)
+        # 目标 6 个月：当月 M 及往回 M-1~M-5
+        target_months = [self.date_info] + [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 6)]
+        # 全历史算完 is_new_high 后只保留这 6 个月再 cache，避免后面 6 次 filter 各自重算全历史窗口
+        df_flag = self.df_new_high_hist.withColumn(
+            'hist_min_before', F.min('rank').over(w)
+        ).withColumn(
+            'is_new_high',
+            F.when(
+                F.col('hist_min_before').isNull() | (F.col('rank') <= F.col('hist_min_before')),
+                F.lit(1)
+            ).otherwise(F.lit(0))
+        ).select('search_term', 'date_info', 'is_new_high') \
+            .filter(F.col('date_info').isin(target_months)) \
+            .repartition(40, 'search_term').cache()
+
+        # 取 M、M-1~M-5 六个月，各 filter+改名，left join 到 df_save；缺月 join 不上填 0
+        # 例：当月 2025-10，某词 M=1、M-1=1、M-2 缺 → this=1,1ago=1,2ago=0 → 近2月都新高命中、近3月不命中
+        df = self.df_save
+        for i in range(6):
+            df_m = df_flag.filter(F.col('date_info') == target_months[i]) \
+                .select('search_term', F.col('is_new_high').alias(self.NEW_HIGH_COLS[i]))
+            df = df.join(df_m, on='search_term', how='left')
+        self.df_save = df.fillna({col_name: 0 for col_name in self.NEW_HIGH_COLS})
+        df_flag.unpersist()
+
+    # 非 (us+month+>=2025-07)：6 个 flag 全填 -1 占位（Java 转 null），保证各分区 schema 一致
+    def handle_new_high_padding(self):
+        for col_name in self.NEW_HIGH_COLS:
+            self.df_save = self.df_save.withColumn(col_name, F.lit(-1))
+
     def handle_365_data(self):
         sql = f"""
             with base_data as (
@@ -499,7 +580,14 @@ class DwtAbaLastChangeRate(object):
             F.col("rank_change_2_month_ago"),
             F.col("rank_change_3_month_ago"),
             F.col("rank_change_4_month_ago"),
-            F.col("rank_change_5_month_ago")
+            F.col("rank_change_5_month_ago"),
+            # 创历史新高 6 flag（1=创新高/0=没创新高/-1=占位），顺序与 Hive ALTER、PG ADD、sqoop_export 严格一致
+            F.col("new_high_this_month"),
+            F.col("new_high_1_month_ago"),
+            F.col("new_high_2_month_ago"),
+            F.col("new_high_3_month_ago"),
+            F.col("new_high_4_month_ago"),
+            F.col("new_high_5_month_ago")
         )
 
         # 类型转换

Pyspark_job/sqoop_export/dwt_aba_last_change_rate.py

@@ -22,7 +22,7 @@ if __name__ == '__main__':
         print("导出到PG测试库中")
     else:
         CommonUtil.judge_is_work_hours(site_name=site_name, date_type=date_type, date_info=date_info,
-                                       principal='chenyuanjie', priority=2, export_tools_type=1,  belonging_to_process=f'新ABA流程_{date_type}')
+                                       principal='hejiangming', priority=2, export_tools_type=1,  belonging_to_process=f'新ABA流程_{date_type}')
         if date_type in (DateTypes.month.name, DateTypes.week.name, DateTypes.month_week.name):
             db_type = 'postgresql_cluster'
             print("导出到PG-Cluster库中")
@@ -121,7 +121,14 @@ if __name__ == '__main__':
             "rank_change_2_month_ago",
             "rank_change_3_month_ago",
             "rank_change_4_month_ago",
-            "rank_change_5_month_ago"
+            "rank_change_5_month_ago",
+            # 创历史新高 6 flag（三站都导；us 真值、uk/de 占位 -1）
+            "new_high_this_month",
+            "new_high_1_month_ago",
+            "new_high_2_month_ago",
+            "new_high_3_month_ago",
+            "new_high_4_month_ago",
+            "new_high_5_month_ago"
         ]
         export_cols = export_base_cols + month_extra_cols
     else: