bsr分类模块-兼容landing_asin判断asin跳转

Pyspark_job/dwt/dwt_bs_top100.py

@@ -141,16 +141,16 @@ class DwtBsTop100(Templates):
         # 【取 node_id：重定向感知映射，不改 dwt_flow】
         # 为什么不直接用 dwt_flow 里的 node：dwt_flow 不带 node 列；若拿 dwt_flow 的 asin 回 join
         #   dim_asin_detail 会踩重定向坑——dwt_flow_asin.py 的 handle_asin_redirect 把 asin 换成了
-        #   current_asin（跳转目标），而 dim_asin_detail 存的是原始 asin，直接 join 会大面积假"缺 node"。
-        # 做法：读 dim 四小列，复刻 dwt_flow 的重定向变换：key=coalesce(current_asin, asin)，
+        #   landing_asin（跳转目标），而 dim_asin_detail 存的是原始 asin，直接 join 会大面积假"缺 node"。
+        # 做法：读 dim 四小列，复刻 dwt_flow 的重定向变换：key=coalesce(landing_asin, asin)，
         #   同 key 按 created_time 降序留一条，得到"重定向后 asin → node_id"，键就能和 dwt_flow 的 asin 对上。
         # ============================================================
-        sql_node = f"""select asin, current_asin, node_id, created_time
+        sql_node = f"""select asin, landing_asin, node_id, created_time
                        from dim_asin_detail
                        where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
         print(f"3. 读取 dim_asin_detail 取 node_id: sql -- {sql_node}")
         df_node = self.spark.sql(sql_node)
-        df_node = df_node.withColumn("asin_eff", F.coalesce(F.col("current_asin"), F.col("asin")))
+        df_node = df_node.withColumn("asin_eff", F.coalesce(F.col("landing_asin"), F.col("asin")))
         w_node = Window.partitionBy("asin_eff").orderBy(F.col("created_time").desc_nulls_last())
         df_node = df_node.withColumn("_rk", F.row_number().over(w_node)).filter(F.col("_rk") == 1)
         # 输出成 asin（重定向后）→ node_id，node_id 转 string 与 keepa cat_id 对齐
@@ -179,7 +179,7 @@ class DwtBsTop100(Templates):
         并复刻 dwt_flow 的跳转（coalesce+created_time 最新去重）对齐到重定向 asin，和 df_bs_data 同 grain。
         """
         # 读 dim 原始 launch 及跳转所需列（与 df_node 同一张 dim，单独取以带上 asin_launch_time）
-        sql_dim = f"""select asin, current_asin, asin_launch_time, created_time
+        sql_dim = f"""select asin, landing_asin, asin_launch_time, created_time
                       from dim_asin_detail
                       where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
         print(f"读取 dim_asin_detail 重算新品: sql -- {sql_dim}")
@@ -202,8 +202,8 @@ class DwtBsTop100(Templates):
         df_dim = df_dim.withColumn(
             "asin_is_new_recalc",
             self.udf_new_asin_flag(F.greatest(F.col('asin_launch_time'), F.col('keepa_tracking_since')), F.lit(cal_date)))
-        # 复刻 dwt_flow 跳转：target=coalesce(current_asin,asin)，同 target 按 created_time 最新留一条
-        df_dim = df_dim.withColumn("asin_eff", F.coalesce(F.col("current_asin"), F.col("asin")))
+        # 复刻 dwt_flow 跳转：target=coalesce(landing_asin,asin)，同 target 按 created_time 最新留一条
+        df_dim = df_dim.withColumn("asin_eff", F.coalesce(F.col("landing_asin"), F.col("asin")))
         w = Window.partitionBy("asin_eff").orderBy(F.col("created_time").desc_nulls_last())
         df_dim = df_dim.withColumn("_rk", F.row_number().over(w)).filter(F.col("_rk") == 1)
         return df_dim.select(F.col("asin_eff").alias("asin"), F.col("asin_is_new_recalc"))
@@ -412,7 +412,7 @@ class DwtBsTop100(Templates):
         print("handle_bs_node_attach 调用（node_id 归属）")
         # ============================================================
         # Step 1：给 asin 挂 node_id
-        # 【为什么 join on asin】df_node 的 key 已按 coalesce(current_asin, asin) 复刻了 dwt_flow 的重定向，
+        # 【为什么 join on asin】df_node 的 key 已按 coalesce(landing_asin, asin) 复刻了 dwt_flow 的重定向，
         #   而 df_bs_data 的 asin 也是 dwt_flow 重定向后的 asin，两边键对得上。
         # 【dwt_flow 是 asin 粒度】一个 asin 一行，join 后仍一个 asin 一行（df_node 也已按 asin 去重）。
         # ============================================================