no message

Pyspark_job/dwt/dwt_aba_last365.py

@@ -331,9 +331,32 @@ class DwtAbaLast365(object):
                     ),
                     ''
                 )
-            """).alias('st_attribute_label')
+            """).alias('st_attribute_label'),
+            # 峰值月中间值：最小 rank（排名最好）+ (rank,月份) 明细数组（下面算 peak_month 用）
+            F.min("rank").alias("min_rank"),
+            F.collect_list(F.struct("rank", "date_info")).alias("rank_arr"),
+            # 常年可卖：近 12 个月都出现在 ABA 月度搜索词中（去重月份数=12）→ 1，否则 0
+            F.when(F.size(F.collect_set("date_info")) >= 12, F.lit(1)).otherwise(F.lit(0)).alias("all_year_text_flag")
         )
 
+        # 峰值月 peak_month：从 rank_arr 里 filter 出 rank=min_rank 的项 → 取 date_info → 排序 → 拼成 'YYYY-MM' 逗号串
+        # 并列峰值全保留；与月表 dwt_aba_st_analytics 同口径同格式
+        # nullif 兜底：极端 rank 全 null 时 filter 为空 → 空串 → 转 null → 由 handle_save 的 na.fill 转空串 → PG string_to_array 转空数组 {}
+        df_agg = df_agg.withColumn(
+            "peak_month",
+            F.expr("""
+                nullif(
+                    concat_ws(',',
+                        array_sort(transform(
+                            filter(rank_arr, x -> x.rank = min_rank),
+                            x -> x.date_info
+                        ))
+                    ),
+                    ''
+                )
+            """)
+        ).drop("min_rank", "rank_arr")
+
         # 行转列的字段
         agg_col_arr = ['st_num', 'bsr_orders', 'orders', 'market_cycle_type', 'search_volume']
         self.df_base = self.pivot_df(
@@ -614,6 +637,10 @@ class DwtAbaLast365(object):
             F.col("appear_month_lastest"),
             #搜索词属性标签(12 个月合并去重,逗号分隔字符串,sqoop 导出 PG 后转 VARCHAR[])
             F.col("st_attribute_label"),
+            # 峰值月：近12个月 rank 最小的月份,并列保留,'YYYY-MM' 逗号分隔(与月表同口径)
+            F.col("peak_month"),
+            # 常年可卖：近12个月每月都出现在 ABA 月度搜索词中=1,否则=0
+            F.col("all_year_text_flag"),
 
             F.lit(self.site_name).alias("site_name"),
             F.lit(self.date_type).alias("date_type"),
@@ -655,7 +682,9 @@ class DwtAbaLast365(object):
             "top_rank": 0,
             # handle_agg 里 12 月全 -1 时输出空串 → nullif 转 null → 这里 fillna '-1'
             # 占位 "-1" 与 dwt_aba_st_analytics 端规则一致,Java 转 null 返前端
-            "st_attribute_label": "-1"
+            "st_attribute_label": "-1",
+            # 峰值月极端兜底（rank 全 null）→ nullif 转 null → 这里 fillna 空串，PG string_to_array 转空数组 {}（与月表同规则）
+            "peak_month": ""
         }).cache()
 
     def save_data(self):

Pyspark_job/dwt/dwt_aba_st_analytics.py

@@ -9,6 +9,7 @@ from pyspark.sql import functions as F
 from pyspark.sql.types import IntegerType
 from utils.db_util import DBUtil
 from utils.spark_util import SparkUtil
+from utils.common_util import CommonUtil
 from yswg_utils.common_udf import udf_detect_phrase_reg
 
 
@@ -57,6 +58,9 @@ class DwtAbaStAnalytics(Templates):
         # 搜索词属性标签 df，来源 dws_st_theme 聚合 theme_ch（材质/颜色/细分人群等）
         # 仅 month 流程在 read_data 阶段真正读取数据；非 month 流程下游统一用 lit("-1") 填充
         self.df_st_attribute = self.spark.sql(f"select 1+1;")
+        # 自身历史 11 个月分区的 rank（峰值月 peak_month / 常年可卖 all_year_text_flag 计算用）
+        # 仅 month 流程在 read_data 阶段真正读取数据；非 month 流程下游统一用占位值填充
+        self.df_history_rank = self.spark.sql(f"select 1+1;")
 
         # 自定义udf函数注册
         self.u_contains = self.spark.udf.register('u_contains', self.udf_contains, IntegerType())
@@ -343,6 +347,13 @@ class DwtAbaStAnalytics(Templates):
         and date_info = '{self.date_info}'
         """
         self.df_st_detail = self.spark.sql(sqlQuery=sql)
+        # ============================================================
+        # 清洗品牌/类目字段里的 \x00（NUL 字节）  脏数据流到本表后，sqoop 导出 PG 会报
+        # 注意：不洗 search_term/asin1-3——search_term 是后续所有 join 的 key，
+        #       单独在这里洗会和其他上游表的 key 对不上，导致数据悄悄丢失
+        # ============================================================
+        for col in ['st_brand1', 'st_brand2', 'st_brand3', 'st_category1', 'st_category2', 'st_category3']:
+            self.df_st_detail = self.df_st_detail.withColumn(col, F.regexp_replace(F.col(col), '\x00', ''))
         self.df_st_detail = self.df_st_detail.repartition(80, 'search_term').cache()
         print("self.df_st_detail:")
         self.df_st_detail.show(10, truncate=True)
@@ -535,6 +546,30 @@ class DwtAbaStAnalytics(Templates):
             print("self.df_st_attribute:")
             self.df_st_attribute.show(10, truncate=True)
 
+            # ============================================================
+            # 读自身历史 11 个月分区的 rank，用于计算峰值月 peak_month / 常年可卖 all_year_text_flag
+            # 【为什么读自己而不读 dim_st_detail】：本表 month 分区从 2024-01 开始齐全，rank 字段
+            #   本身就来源于 dim_st_detail，读自己只需 3 列、无需兼容 month_old 老分区格式
+            # 【为什么只读 11 个月】：当月分区此时还没写入（本任务正在算），当月 rank
+            #   在 handle_peak_month 阶段直接从 df_save 里取，再和这 11 个月 union 凑满 12 个月窗口
+            # 【边界】：窗口内某月分区缺失（如词库早期月份），in 条件查不到就少一个月，
+            #   不影响 min(rank) 取峰值，只影响 all_year_text_flag（不满 12 个月必然=0，符合业务语义）
+            # ============================================================
+            last_11_month = [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 12)]
+            sql = f"""
+            select
+                search_term,
+                rank,
+                date_info
+            from dwt_aba_st_analytics
+            where site_name = '{self.site_name}'
+              and date_type = 'month'
+              and date_info in ({CommonUtil.list_to_insql(last_11_month)})
+            """
+            self.df_history_rank = self.spark.sql(sqlQuery=sql).repartition(80, 'search_term').cache()
+            print("self.df_history_rank:")
+            self.df_history_rank.show(10, truncate=True)
+
     def handle_data(self):
         # 对基础计算表进行关联
         self.handle_base_join()
@@ -553,8 +588,15 @@ class DwtAbaStAnalytics(Templates):
 
         if self.date_type == 'month':
             self.handle_first_ever_flag() # 全历史首次出现标记（基于累加表 dim_st_detail_history）  不是月流程填充 -1
+            self.handle_peak_month()      # 峰值月 + 常年可卖（基于自身历史 11 个月分区 + 当月 rank）
         else:
             self.df_save = self.df_save.withColumn('is_first_ever_text', F.lit(-1))
+            # 非 month 流程业务不需要峰值月/常年可卖，直接占位（peak_month 空串→PG 转空数组 {}，all_year_text_flag 占位 -1）
+            self.df_save = self.df_save.withColumn(
+                'peak_month', F.lit('')
+            ).withColumn(
+                'all_year_text_flag', F.lit(-1)
+            )
 
         # 附加属性标签字段 st_attribute_label
         # month 流程：left join read_data 阶段已读取的 dws_st_theme 聚合结果，join 不上的词在 handle_column 阶段 fillna("-1")
@@ -919,6 +961,69 @@ class DwtAbaStAnalytics(Templates):
                                    .fillna({'is_first_ever_text': 1})
         self.df_history_st.unpersist()
 
+    # 峰值月 + 常年可卖：基于近 12 个月（自身历史 11 个月分区 + 当月）的 rank 计算
+    def handle_peak_month(self):
+        """
+        peak_month：近 12 个月窗口内 rank 最小（排名最好）的月份，并列全保留，
+                    按时间升序逗号拼接，格式 'YYYY-MM'（如 '2025-11,2026-04'）。
+        all_year_text_flag：近 12 个月每个月都出现在 ABA 月度搜索词中 → 1，否则 0（常年可卖）。
+        """
+        # ============================================================
+        # Step 1：拼出完整 12 个月的 (search_term, rank, date_info) 明细
+        # 【为什么当月 rank 从 df_save 取】：当月分区还没写入 Hive，读不到自己；
+        #   df_save 此时已含当月 rank（来源 dim_st_detail 的 st_rank，inner join 必有值）
+        # 【例子】：词 'water bottle'，历史 11 个月出现 5 次 + 当月 1 行 → union 后 6 行
+        # ============================================================
+        df_current_rank = self.df_save.select(
+            'search_term', 'rank', F.lit(self.date_info).alias('date_info')
+        )
+        df_rank_all = self.df_history_rank.unionByName(df_current_rank)
+
+        # ============================================================
+        # Step 2：单次 groupBy 一并算出 最小 rank / (rank,月份)明细数组 / 去重月份数
+        # 【为什么不用 window 再 groupBy】：window.partitionBy('search_term') 和
+        #   groupBy('search_term') 会按同一个 key 各洗一次 shuffle（df_rank_all ≈ 2000w 行），
+        #   把 min_rank 合进 groupBy 一次算完，省掉 window 那次 shuffle
+        # 【例子】：'water bottle' → min_rank=100, rank_arr=[(100,'2025-11'),(100,'2026-04'),(300,..)], month_cnt=6
+        # ============================================================
+        df_agg = df_rank_all.groupBy('search_term').agg(
+            F.min('rank').alias('min_rank'),
+            F.collect_list(F.struct('rank', 'date_info')).alias('rank_arr'),
+            F.count_distinct('date_info').alias('month_cnt')
+        )
+
+        # ============================================================
+        # Step 3：map 端表达式算出 peak_month / all_year_text_flag（不触发 shuffle）
+        # 【peak_month 怎么算】：从 rank_arr 里 filter 出 rank=min_rank 的项 → transform 取 date_info
+        #   → array_sort 保证时间升序、重跑结果稳定 → concat_ws 拼成逗号串
+        # 【例子】：min_rank=100，命中 '2025-11'、'2026-04' → peak_month='2025-11,2026-04'
+        # 【all_year_text_flag 怎么算】：去重月份数 = 12 → 常年可卖；只出现 6 个月 → 0
+        # 【nullif 兜底】：极端 rank 全 null（理论不会发生，rank 来源 inner join）时 filter 为空 →
+        #   concat_ws 输出空串 '' → nullif 转 null → 由 handle_column 的 na.fill 转空串 → PG string_to_array 转空数组 {}
+        # ============================================================
+        df_peak = df_agg.select(
+            'search_term',
+            F.expr("""
+                nullif(
+                    concat_ws(',',
+                        array_sort(transform(
+                            filter(rank_arr, x -> x.rank = min_rank),
+                            x -> x.date_info
+                        ))
+                    ),
+                    ''
+                )
+            """).alias('peak_month'),
+            F.when(F.col('month_cnt') >= 12, F.lit(1)).otherwise(F.lit(0)).alias('all_year_text_flag')
+        )
+
+        # ============================================================
+        # Step 4：join 回 df_save
+        # 【为什么 left join 也必命中】：df_rank_all 包含当月全部词，df_peak 必有每个当月词
+        # ============================================================
+        self.df_save = self.df_save.join(df_peak, on='search_term', how='left')
+        self.df_history_rank.unpersist()
+
     def handle_column(self):
         # 入库前字段处理，精度以 PG 集群列约束为基准
         self.df_save = self.df_save.select(
@@ -1030,7 +1135,9 @@ class DwtAbaStAnalytics(Templates):
             "is_first_ever_text",       # 需求2：全历史首次出现标记（1=当月首次, 0=历史出现过）
             "brand_asin_proportion",    # 需求3：前三页ASIN数最多品牌的ASIN数占比
             "seller_asin_proportion",   # 需求3：前三页ASIN数最多卖家的ASIN数占比
-            "st_attribute_label"        # 搜索词属性标签（逗号分隔字符串，sqoop 导出 PG 后转 VARCHAR[]）
+            "st_attribute_label",       # 搜索词属性标签（逗号分隔字符串，sqoop 导出 PG 后转 VARCHAR[]）
+            "peak_month",               # 峰值月：近12个月 rank 最小的月份，并列保留，'YYYY-MM' 逗号分隔（PG 转 VARCHAR[]）
+            "all_year_text_flag"          # 常年可卖：近12个月每月都出现在 ABA 月度搜索词中=1，否则=0
         )
 
         # 空值处理
@@ -1049,7 +1156,9 @@ class DwtAbaStAnalytics(Templates):
             "is_hidden_cate": 0,
             "brand_asin_proportion": -1,   # 需求3：分子为 null（搜索词全无品牌） → 占位 -1
             "seller_asin_proportion": -1,  # 需求3：分子为 null（搜索词全无账号） → 占位 -1
-            "st_attribute_label": "-1"     # 词典无匹配 → 占位 "-1"（Java 侧转 null 返前端）
+            "st_attribute_label": "-1",    # 词典无匹配 → 占位 "-1"（Java 侧转 null 返前端）
+            "peak_month": "",              # 极端兜底（rank 全 null）→ 空串，PG string_to_array 转空数组 {}
+            "all_year_text_flag": 0          # 理论必有值，兜底 0（非 month 流程已提前填 -1，不受影响）
         })
 
         # 日期字段补全

Pyspark_job/sqoop_export/dwt_aba_last365.py

@@ -46,6 +46,7 @@ if __name__ == '__main__':
             -- st_attribute_label 在正式表是 VARCHAR[]，Sqoop 无法直接写入数组类型
             -- 先临时改成 VARCHAR(200) 让 Sqoop 写字符串，交换完成后再 ALTER 回 VARCHAR[]
             ALTER TABLE {export_tb_rel} ALTER COLUMN st_attribute_label TYPE VARCHAR(200);
+            ALTER TABLE {export_tb_rel} ALTER COLUMN peak_month TYPE VARCHAR(200);
         """
         print("================================执行sql================================")
         print(sql)
@@ -192,7 +193,11 @@ if __name__ == '__main__':
             "rank_change_rate_lastest",
             "rank_rate_of_change_lastest",
             # 搜索词属性标签(12 个月合并去重,逗号分隔字符串,sqoop 导出 PG 后转 VARCHAR[])
-            "st_attribute_label"
+            "st_attribute_label",
+            # 峰值月(近12月rank最小的月份,逗号分隔字符串,sqoop 导出 PG 后转 VARCHAR[])
+            "peak_month",
+            # 常年可卖标记(标量 int,sqoop 直写)
+            "all_year_text_flag"
         ],
         partition_dict={
             "site_name": site_name,
@@ -227,6 +232,10 @@ if __name__ == '__main__':
             ALTER TABLE {export_tb_before}
             ALTER COLUMN st_attribute_label TYPE VARCHAR[]
             USING string_to_array(st_attribute_label, ',')::varchar[];
+            
+            ALTER TABLE {export_tb_before}
+            ALTER COLUMN peak_month TYPE VARCHAR[]
+            USING string_to_array(coalesce(peak_month, ''), ',')::varchar[];
 
             alter table {export_tb_before} drop if exists keyword_tsv;
             alter table {export_tb_before} add column keyword_tsv tsvector generated always as (to_tsvector('english_amazonword', search_term)) STORED;

Pyspark_job/sqoop_export/dwt_aba_st_analytics.py

@@ -186,16 +186,18 @@ if __name__ == '__main__':
             tb_cols = [
                 "is_new_market_segment", "color_proportion", "supply_demand", "market_cycle_type", "is_high_return_text",
                 "st_zr_counts", "st_sp_counts", "st_self_asin_counts", "st_self_asin_proportion",
-                # 需求2 + 需求3：月度专属字段（仅 month 流程才有意义）
-                # is_first_ever_text 依赖累加表 dim_st_detail_history（仅 month 数据）
-                # brand_asin_proportion / seller_asin_proportion 服务月搜索词筛选页面
                 "is_first_ever_text",
                 "brand_asin_proportion",
                 "seller_asin_proportion",
                 # 搜索词属性标签（材质/颜色/细分人群等），仅 month 计算
                 # Hive 端是逗号分隔 STRING（如 "材质,颜色"），sqoop 写入 PG copy 表需先 ALTER 成 VARCHAR
                 # 交换前再 ALTER 回 VARCHAR[]（用 string_to_array 转换），与 dwt_aba_last365 处理 st_movie_brand_label 同款思路
-                "st_attribute_label"
+                "st_attribute_label",
+                # 峰值月：Hive 端是逗号分隔 STRING（如 "2025-11,2026-04"），PG 端是 VARCHAR[]
+                # 同 st_attribute_label：copy 表先 ALTER 成 VARCHAR 让 sqoop 写字符串，交换前再 string_to_array 转回 VARCHAR[]
+                "peak_month",
+                # 常年可卖标记：标量 int，sqoop 直写，无需中转
+                "all_year_text_flag"
             ]
             # 处理导出表
             export_master_tb = f"{export_base_tb}_{date_type}_{year_str}"
@@ -237,9 +239,9 @@ if __name__ == '__main__':
             # copy 表继承自正式分区表（含 st_attribute_label VARCHAR[]），
             # 但 Sqoop 不支持直接写入 PG 数组类型，必须先把 copy 表的该列临时改成 VARCHAR
             # 等 Sqoop 完成后、分区交换之前，再 ALTER 回 VARCHAR[]（见下方 exchange_pg_part_tb 前的处理）
-            # 这是与 sqoop_export/dwt_aba_last365.py 中 st_movie_brand_label 同款的"VARCHAR 中转"模式
             sql_alter_to_varchar = f"""
                   ALTER TABLE {export_tb_copy} ALTER COLUMN st_attribute_label TYPE VARCHAR(200);
+                  ALTER TABLE {export_tb_copy} ALTER COLUMN peak_month TYPE VARCHAR(200);
                   """
             DBUtil.engine_exec_sql(engine, sql_alter_to_varchar)
 
@@ -250,8 +252,6 @@ if __name__ == '__main__':
             #     "column keyword_tsv in child table must be a generated column"
             # 解决：显式 drop 后以生成列方式重建。CASCADE 是为了同时把 LIKE 时附带继承的 keyword_tsv 索引一并清掉，
             # 后面 ATTACH 时 master 的分区索引会自动给新分区补建对应索引。
-            # 重建在 Sqoop 之前做：生成列由 search_term 自动计算（STORED），Sqoop 不写也不会报错。
-            # 参考 sqoop_export/dwt_aba_last365.py 中对 keyword_tsv 的同款处理。
             sql_fix_keyword_tsv = f"""
                   ALTER TABLE {export_tb_copy} DROP COLUMN IF EXISTS keyword_tsv CASCADE;
                   ALTER TABLE {export_tb_copy} ADD COLUMN keyword_tsv tsvector
@@ -311,12 +311,15 @@ if __name__ == '__main__':
     elif date_type == DateTypes.month.name:
         # 分区交换前必须把 copy 表的 st_attribute_label 从 VARCHAR 转回 VARCHAR[]
         # 否则与 master 分区表 schema 不一致，exchange_pg_part_tb 会失败
-        # USING string_to_array(...) 把 Sqoop 写入的逗号串（如 "材质,颜色"）拆成数组（如 {材质,颜色}）
-        # 词典无匹配的词，PySpark 已 fillna 为 "-1"，转换后是 {-1}，与 Java 占位约定一致
+        # USING string_to_array(...) 把 Sqoop 写入的逗号串（如 "材质,颜色"）拆成数组（如 {材质,颜色}
         sql_alter_back = f"""
               ALTER TABLE {export_tb_copy}
               ALTER COLUMN st_attribute_label TYPE VARCHAR[]
               USING string_to_array(st_attribute_label, ',')::varchar[];
+
+              ALTER TABLE {export_tb_copy}
+              ALTER COLUMN peak_month TYPE VARCHAR[]
+              USING string_to_array(coalesce(peak_month, ''), ',')::varchar[];
               """
         DBUtil.engine_exec_sql(engine, sql_alter_back)