asin跳转逻辑优化

Pyspark_job/dwd/dwd_st_asin_measure.py

@@ -5,6 +5,7 @@ import sys
 sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
 from pyspark.storagelevel import StorageLevel
 from utils.templates import Templates
+from yswg_utils.common_udf import resolve_asin_jump
 # from ..utils.templates import Templates
 # from AmazonSpider.pyspark_job.utils.templates_test import Templates
 from pyspark.sql.types import StringType, IntegerType
@@ -177,7 +178,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, updated_time, landing_asin " \
+        sql = f"select asin, asin_title, asin_price, parent_asin, asin_bought_month, updated_time, current_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 +362,21 @@ class DwdStMeasure(Templates):
             self.df_st_asin_flow, on=['page_rank'], how='left'
         )
 
-        # === asin跳转处理：将df_st_asin和df_asin_detail的asin统一替换为landing_asin ===
-        # 1. 保留原始缓存引用，从中构建跳转映射：原始asin → landing_asin
+        # === asin跳转处理：判断规则见 resolve_asin_jump，
+        # 将df_st_asin和df_asin_detail的asin统一替换 ===
+        # 1. 保留原始缓存引用，从中构建跳转映射：原始asin → current_asin
         df_asin_detail_cached = self.df_asin_detail
-        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（无跳转则保留原值）
+        redirect_map = df_asin_detail_cached.select(F.col('asin').alias('redirect_src'), F.col('current_asin'))
+        # 2. df_st_asin中的原始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('landing_asin'), F.col('asin'))) \
-            .drop('redirect_src', 'landing_asin')
+            .withColumn('asin', resolve_asin_jump()) \
+            .drop('redirect_src', 'current_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('landing_asin'), F.col('asin'))) \
-            .drop('landing_asin') \
+            .withColumn('asin', resolve_asin_jump()) \
+            .drop('current_asin') \
             .withColumn('_rk', F.row_number().over(window_redirect)) \
             .filter(F.col('_rk') == 1) \
             .drop('_rk', 'updated_time') \

Pyspark_job/dwt/dwt_bs_top100.py

@@ -16,7 +16,7 @@ from pyspark.storagelevel import StorageLevel
 from utils.templates import Templates
 from utils.hdfs_utils import HdfsUtils          # 落表前清 HDFS 分区目录（幂等重跑）
 from utils.common_util import CommonUtil        # 构造 HDFS 分区路径
-from yswg_utils.common_udf import udf_new_asin_flag  # 新品判断 UDF（与 dim 同一个，用于按新规则重算 asin_is_new）
+from yswg_utils.common_udf import udf_new_asin_flag, resolve_asin_jump  # 新品判断UDF + asin跳转判断
 # from ..utils.templates import Templates
 # from AmazonSpider.pyspark_job.utils.templates_test import Templates
 from pyspark.sql.types import StringType, IntegerType, DoubleType, LongType
@@ -141,16 +141,17 @@ 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 换成了
-        #   landing_asin（跳转目标），而 dim_asin_detail 存的是原始 asin，直接 join 会大面积假"缺 node"。
-        # 做法：读 dim 四小列，复刻 dwt_flow 的重定向变换：key=coalesce(landing_asin, asin)，
+        #   current_asin（跳转目标，判断规则见 resolve_asin_jump），而 dim_asin_detail
+        #   存的是原始 asin，直接 join 会大面积假"缺 node"。
+        # 做法：读 dim 四小列，复刻 dwt_flow 的重定向变换：key=resolve_asin_jump(asin, current_asin)，
         #   同 key 按 created_time 降序留一条，得到"重定向后 asin → node_id"，键就能和 dwt_flow 的 asin 对上。
         # ============================================================
-        sql_node = f"""select asin, landing_asin, node_id, created_time
+        sql_node = f"""select asin, current_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("landing_asin"), F.col("asin")))
+        df_node = df_node.withColumn("asin_eff", resolve_asin_jump())
         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 +180,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, landing_asin, asin_launch_time, created_time
+        sql_dim = f"""select asin, current_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 +203,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(landing_asin,asin)，同 target 按 created_time 最新留一条
-        df_dim = df_dim.withColumn("asin_eff", F.coalesce(F.col("landing_asin"), F.col("asin")))
+        # 复刻 dwt_flow 跳转：target=resolve_asin_jump(asin, current_asin)，同 target 按 created_time 最新留一条
+        df_dim = df_dim.withColumn("asin_eff", resolve_asin_jump())
         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 +413,7 @@ class DwtBsTop100(Templates):
         print("handle_bs_node_attach 调用（node_id 归属）")
         # ============================================================
         # Step 1：给 asin 挂 node_id
-        # 【为什么 join on asin】df_node 的 key 已按 coalesce(landing_asin, asin) 复刻了 dwt_flow 的重定向，
+        # 【为什么 join on asin】df_node 的 key 已按 resolve_asin_jump 复刻了 dwt_flow 的重定向，
         #   而 df_bs_data 的 asin 也是 dwt_flow 重定向后的 asin，两边键对得上。
         # 【dwt_flow 是 asin 粒度】一个 asin 一行，join 后仍一个 asin 一行（df_node 也已按 asin 去重）。
         # ============================================================

Pyspark_job/dwt/dwt_flow_asin.py

@@ -31,6 +31,7 @@ from utils.hdfs_utils import HdfsUtils
 from pyspark.storagelevel import StorageLevel
 from pyspark.sql.types import *
 from utils.DorisHelper import DorisHelper
+from yswg_utils.common_udf import resolve_asin_jump
 
 
 class DwtFlowAsin(Templates):
@@ -212,7 +213,7 @@ class DwtFlowAsin(Templates):
             date_format(created_time, 'yyyy-MM-dd HH:mm:ss') as asin_crawl_date, asin_bought_month, asin_image_view,  
             case when product_description is not null then 1 else 0 end as is_with_product_description, asin_describe, 
             category_id as top_category_id, category_first_id as top_category_first_id, customer_reviews_json, img_list as img_info, 
-            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, variat_list, landing_asin
+            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, variat_list, current_asin
             from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
         print("sql:" + sql)
         self.df_asin_detail = self.spark.sql(sqlQuery=sql)
@@ -888,12 +889,12 @@ class DwtFlowAsin(Templates):
                 print("不用导出旧数据到doris中")
 
     def handle_asin_redirect(self):
-        """asin跳转处理：将跳转asin替换为landing_asin，按asin去重保留最新抓取时间"""
+        """asin跳转处理：判断规则见 resolve_asin_jump；按asin去重保留最新抓取时间"""
         df_old = self.df_asin_detail
         window_redirect = Window.partitionBy('asin').orderBy(F.col('asin_crawl_date').desc_nulls_last())
         self.df_asin_detail = df_old \
-            .withColumn('asin', F.coalesce(F.col('landing_asin'), F.col('asin'))) \
-            .drop('landing_asin') \
+            .withColumn('asin', resolve_asin_jump()) \
+            .drop('current_asin') \
             .withColumn('_rk', F.row_number().over(window_redirect)) \
             .filter(F.col('_rk') == 1) \
             .drop('_rk') \

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -18,7 +18,7 @@ from pyspark.sql import Window
 from pyspark.storagelevel import StorageLevel
 from utils.DorisHelper import DorisHelper
 from yswg_utils.common_df import get_node_first_id_df, get_first_id_from_category_desc_df
-from yswg_utils.common_udf import udf_parse_bs_category, parse_weight_str, udf_extract_volume_dimensions, udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders
+from yswg_utils.common_udf import udf_parse_bs_category, parse_weight_str, udf_extract_volume_dimensions, udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders, resolve_asin_jump
 
 
 class KafkaFlowAsinDetail(Templates):
@@ -182,27 +182,15 @@ class KafkaFlowAsinDetail(Templates):
             StructField("follow_sellers", IntegerType(), True),
             StructField("fbm_delivery_price", FloatType(), True),
             StructField("product_json", StringType(), True),
-            StructField("amazon_label", StringType(), True),
-            StructField("landing_asin", StringType(), True)
+            StructField("amazon_label", StringType(), True)
         ])
         return schema
 
-    # 0. 处理跳转asin：landing_asin 非空时用其替换 asin（在去重前执行，保证跳转后的重复 asin 能被正确合并）
-    # current_asin 保留在df里不做处理，随数据一起落地，方便后续排查跳转是否准确
-    def handle_asin_jump(self, df):
-        df = df.withColumn(
-            "asin",
-            F.when(
-                F.col("landing_asin").isNotNull() & (F.col("landing_asin") != ""),
-                F.col("landing_asin")
-            ).otherwise(F.col("asin"))
-        ).drop("landing_asin")
-        return df
-
-    # 覆写模板去重方法：去重前先做 landing_asin 替换，确保跳转后同 asin 多条记录在去重时被合并
+    # 覆写模板去重方法：去重前先处理跳转asin（判断规则见 resolve_asin_jump），
+    # 确保跳转后同 asin 多条记录在去重时被合并（在去重前执行）
     def deduplication_kafka_data(self, kafka_df, deduplicaiton_key_field, deduplication_time_field):
-        if deduplicaiton_key_field == "asin" and "landing_asin" in kafka_df.columns:
-            kafka_df = self.handle_asin_jump(kafka_df)
+        if deduplicaiton_key_field == "asin" and "current_asin" in kafka_df.columns:
+            kafka_df = kafka_df.withColumn("asin", resolve_asin_jump())
         return super().deduplication_kafka_data(kafka_df, deduplicaiton_key_field, deduplication_time_field)
 
     # 1. 处理asin分类及排名以及排名类型字段