流量选品-月/周隐藏分类修复

Pyspark_job/dwt/dwt_flow_asin.py

@@ -33,6 +33,7 @@ from pyspark.storagelevel import StorageLevel
 from pyspark.sql.types import *
 from utils.DorisHelper import DorisHelper
 from yswg_utils.common_udf import resolve_asin_jump
+from yswg_utils.common_df import get_first_id_from_category_desc_df
 
 
 class DwtFlowAsin(Templates):
@@ -639,12 +640,11 @@ class DwtFlowAsin(Templates):
             "desc_category_first_name",
             F.lower(F.trim(F.split(F.col("asin_category_desc"), "›").getItem(0)))
         )
-        # 读取 Hive 分类维表，获取分类名称与ID的对应关系
-        sql_dim = f"""
-            select lower(trim(en_name)) as desc_category_first_name, category_first_id as desc_category_first_id
-            from dim_bsr_category_tree where site_name = '{self.site_name}' and category_parent_id = 0 and leaf_node = 2
-        """
-        df_bsr_category = F.broadcast(self.spark.sql(sqlQuery=sql_dim))
+        # 获取一级分类名称与ID的对应关系（与30天流程一致）
+        df_bsr_category = F.broadcast(get_first_id_from_category_desc_df(self.site_name, self.spark).select(
+            F.lower(F.col('category_first_name')).alias('desc_category_first_name'),
+            F.col('category_first_id').alias('desc_category_first_id')
+        ))
         # join 补充分类ID
         self.df_asin_detail = self.df_asin_detail.join(df_bsr_category, on=['desc_category_first_name'], how='left')
         # 两个分类ID均在过滤列表中才标记为1

Pyspark_job/dwt/dwt_flow_asin_week.py

@@ -32,6 +32,7 @@ from utils.hdfs_utils import HdfsUtils
 from pyspark.storagelevel import StorageLevel
 from pyspark.sql.types import *
 from yswg_utils.common_udf import resolve_asin_jump
+from yswg_utils.common_df import get_first_id_from_category_desc_df
 
 
 class DwtFlowAsinWeek(Templates):
@@ -564,11 +565,11 @@ class DwtFlowAsinWeek(Templates):
             "desc_category_first_name",
             F.lower(F.trim(F.split(F.col("asin_category_desc"), "›").getItem(0)))
         )
-        sql_dim = f"""
-            select lower(trim(en_name)) as desc_category_first_name, category_first_id as desc_category_first_id
-            from dim_bsr_category_tree where site_name = '{self.site_name}' and category_parent_id = 0 and leaf_node = 2
-        """
-        df_bsr_category = F.broadcast(self.spark.sql(sqlQuery=sql_dim))
+        # 获取一级分类名称与ID的对应关系（与30天流程一致）
+        df_bsr_category = F.broadcast(get_first_id_from_category_desc_df(self.site_name, self.spark).select(
+            F.lower(F.col('category_first_name')).alias('desc_category_first_name'),
+            F.col('category_first_id').alias('desc_category_first_id')
+        ))
         self.df_asin_detail = self.df_asin_detail.join(df_bsr_category, on=['desc_category_first_name'], how='left')
         # 两个分类ID均在过滤列表中才标记为1
         need_categories = "('mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text', 'magazines', 'movies-tv', 'software', 'videogames', 'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed')"