流量选品30天模块-兼容landing_asin字段判断asin跳转

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -182,24 +182,26 @@ 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("amazon_label", StringType(), True),
+            StructField("landing_asin", StringType(), True)
         ])
         return schema
 
-    # 0. 处理跳转asin：current_asin 非空时用其替换 asin（在去重前执行，保证跳转后的重复 asin 能被正确合并）
+    # 0. 处理跳转asin：landing_asin 非空时用其替换 asin（在去重前执行，保证跳转后的重复 asin 能被正确合并）
+    # current_asin 保留在df里不做处理，随数据一起落地，方便后续排查跳转是否准确
     def handle_asin_jump(self, df):
         df = df.withColumn(
             "asin",
             F.when(
-                F.col("current_asin").isNotNull() & (F.col("current_asin") != ""),
-                F.col("current_asin")
+                F.col("landing_asin").isNotNull() & (F.col("landing_asin") != ""),
+                F.col("landing_asin")
             ).otherwise(F.col("asin"))
-        ).drop("current_asin")
+        ).drop("landing_asin")
         return df
 
-    # 覆写模板去重方法：去重前先做 current_asin 替换，确保跳转后同 asin 多条记录在去重时被合并
+    # 覆写模板去重方法：去重前先做 landing_asin 替换，确保跳转后同 asin 多条记录在去重时被合并
     def deduplication_kafka_data(self, kafka_df, deduplicaiton_key_field, deduplication_time_field):
-        if deduplicaiton_key_field == "asin" and "current_asin" in kafka_df.columns:
+        if deduplicaiton_key_field == "asin" and "landing_asin" in kafka_df.columns:
             kafka_df = self.handle_asin_jump(kafka_df)
         return super().deduplication_kafka_data(kafka_df, deduplicaiton_key_field, deduplication_time_field)