搜索量同比增长列 相关三个字段的计算从搜索量改为排名

Pyspark_job/dwt/dwt_aba_last365.py

@@ -291,16 +291,58 @@ class DwtAbaLast365(object):
         self.df_last_year = self.spark.sql(sql).repartition(80, 'search_term').cache()
 
         # 搜索量同比增长识别:判断持续上升/下降
+        # ===== 原逻辑(保留备查):用 search_volume_change_rate 判断搜索量同比趋势 =====
+        # 问题:search_volume 来自 ods_rank_search_rate_repeat(rank→search_num 映射表),
+        #      该表从 2023-03 后停更 → 所有月份用同一份快照 → 同排名的词 search_volume 不变 → 同比=0,失效。
+        # 改用 rank_change_rate(ABA 排名同比变化率):rank 每月是真实值,不依赖停更的映射表。
+        # 符号方向:rank 越小越好,所以 rank_change_rate < 0 = 排名上升 = rising, > 0 = 排名下降 = decline。
+        # 读近12月已有的排名同比变化率(只包含当月在榜的词),额外带 date_info 用于下面 anti join
         sql = f"""
             select
                 search_term,
-                search_volume_change_rate 
+                rank_change_rate,
+                date_info
             from dwt_aba_last_change_rate
             where site_name = '{self.site_name}'
               and date_type = '{self.date_type_original}'
               and date_info in ({CommonUtil.list_to_insql(self.last_12_month)});
         """
-        self.df_sv_change_rate = self.spark.sql(sql).repartition(80, 'search_term').cache()
+        df_sv_raw = self.spark.sql(sql).repartition(80, 'search_term').cache()
+
+        # 补"下榜月"：去年同月在榜、今年该月不在榜 → 补 rank_change_rate = +1000(排名变大=下降)
+        # 和 dwt_aba_last_change_rate 里新进榜的 na.fill(-1000)(排名变小=上升) 对称;
+        # 原逻辑缺少这一面 → 下榜月被跳过 → sv_decline_rate 偏低
+        # 1. 读去年同期12个月哪些词在 dim_st_detail 中存在
+        last_year_months = [CommonUtil.get_month_offset(m, -12) for m in self.last_12_month]
+        sql_ly = f"""
+            select distinct search_term, date_info
+            from dim_st_detail
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type_original}'
+              and date_info in ({CommonUtil.list_to_insql(last_year_months)})
+              and st_rank > 0
+        """
+        df_last_year_presence = self.spark.sql(sql_ly)
+        # 2. 去年月 +12 映射到今年对应月(如 2024-07 → 2025-07)
+        df_last_year_mapped = df_last_year_presence.withColumn(
+            'date_info',
+            F.date_format(F.add_months(F.to_date(F.concat(F.col('date_info'), F.lit('-01'))), 12), 'yyyy-MM')
+        )
+        # 3. anti join 找下榜月:去年有但今年 change_rate 表没有的(词,月)
+        df_sv_existing = df_sv_raw.select('search_term', 'date_info').dropDuplicates()
+        df_dropped = df_last_year_mapped.join(df_sv_existing, on=['search_term', 'date_info'], how='left_anti')
+        # 4. 只保留 df_base 里有的词(完全不在年表的词不需要补)
+        df_base_terms = self.df_base.select('search_term').dropDuplicates()
+        df_dropped = df_dropped.join(df_base_terms, on='search_term', how='inner')
+        # 5. 补 +1000(排名变大=下降,和新进榜-1000对称) 并 union 回原数据
+        df_dropped_sv = df_dropped.select(
+            'search_term', F.lit(1000.0).alias('rank_change_rate')
+        )
+        self.df_sv_change_rate = df_sv_raw.select(
+            'search_term', 'rank_change_rate'
+        ).union(df_dropped_sv).repartition(80, 'search_term').cache()
+        self.df_sv_change_rate.count()  # 触发物化,之后释放中间 cache
+        df_sv_raw.unpersist()
 
         # 影视+品牌标签识别
         sql = f"""
@@ -523,11 +565,12 @@ class DwtAbaLast365(object):
         fields = fields_first_round + fields_second_round
         self.df_base = self.df_base.withColumn('market_cycle_type', F.coalesce(*fields))
 
-        # 持续上升、下降判断
+        # 持续上升、下降判断(用 rank_change_rate,符号方向与 search_volume 相反:
+        #   rank 越小越好,所以 < 0 = 排名上升 = rising, > 0 = 排名下降 = decline)
         self.df_sv_change_rate = self.df_sv_change_rate.withColumn(
-            'sv_rising_flag', F.when(F.col('search_volume_change_rate') > 0, 1).otherwise(0)
+            'sv_rising_flag', F.when(F.col('rank_change_rate') < 0, 1).otherwise(0)
         ).withColumn(
-            'sv_decline_flag', F.when(F.col('search_volume_change_rate') < 0, 1).otherwise(0)
+            'sv_decline_flag', F.when(F.col('rank_change_rate') > 0, 1).otherwise(0)
         )
         # # 计算上升率、下降率
         self.df_sv_change_rate = self.df_sv_change_rate.groupBy('search_term').agg(
@@ -549,7 +592,7 @@ class DwtAbaLast365(object):
 
     def handle_calc_lang(self):
         sql = """
-            select word, langs from big_data_selection.tmp_lang_word_frequency;
+            select word, langs from big_data_selection.lang_word_frequency;
         """
         lang_word_list = self.spark.sql(sql).collect()
         # 转为map