搜索词中间表兼容landing_asin字段做asin跳转

Pyspark_job/dwd/dwd_st_asin_measure.py

@@ -177,7 +177,7 @@ class DwdStMeasure(Templates):
         self.df_asin_bs = self.spark.sql(sql).cache()
         self.df_asin_bs.show(10)
 
-        sql = f"select asin, asin_title, asin_price, parent_asin, asin_bought_month, current_asin, updated_time " \
+        sql = f"select asin, asin_title, asin_price, parent_asin, asin_bought_month, updated_time, landing_asin " \
               f"from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type.replace('_old', '')}' and date_info='{self.date_info}';"
         print("sql:", sql)
         self.df_asin_detail = self.spark.sql(sql).cache()
@@ -361,21 +361,21 @@ class DwdStMeasure(Templates):
             self.df_st_asin_flow, on=['page_rank'], how='left'
         )
 
-        # === asin跳转处理：将df_st_asin和df_asin_detail的asin统一替换为current_asin ===
-        # 1. 保留原始缓存引用，从中构建跳转映射：原始asin → current_asin
+        # === asin跳转处理：将df_st_asin和df_asin_detail的asin统一替换为landing_asin ===
+        # 1. 保留原始缓存引用，从中构建跳转映射：原始asin → landing_asin
         df_asin_detail_cached = self.df_asin_detail
-        redirect_map = df_asin_detail_cached.filter(F.col('current_asin').isNotNull()) \
-            .select(F.col('asin').alias('redirect_src'), F.col('current_asin'))
-        # 2. df_st_asin中的原始asin替换为current_asin（无跳转则保留原值）
+        redirect_map = df_asin_detail_cached.filter(F.col('landing_asin').isNotNull()) \
+            .select(F.col('asin').alias('redirect_src'), F.col('landing_asin'))
+        # 2. df_st_asin中的原始asin替换为landing_asin（无跳转则保留原值）
         self.df_st_asin = self.df_st_asin \
             .join(redirect_map, self.df_st_asin.asin == redirect_map.redirect_src, how='left') \
-            .withColumn('asin', F.coalesce(F.col('current_asin'), F.col('asin'))) \
-            .drop('redirect_src', 'current_asin')
+            .withColumn('asin', F.coalesce(F.col('landing_asin'), F.col('asin'))) \
+            .drop('redirect_src', 'landing_asin')
         # 3. df_asin_detail的asin键同步替换，asin重复时按updated_time降序保留最新一条；释放原始缓存
         window_redirect = Window.partitionBy('asin').orderBy(F.col('updated_time').desc())
         self.df_asin_detail = df_asin_detail_cached \
-            .withColumn('asin', F.coalesce(F.col('current_asin'), F.col('asin'))) \
-            .drop('current_asin') \
+            .withColumn('asin', F.coalesce(F.col('landing_asin'), F.col('asin'))) \
+            .drop('landing_asin') \
             .withColumn('_rk', F.row_number().over(window_redirect)) \
             .filter(F.col('_rk') == 1) \
             .drop('_rk', 'updated_time') \