流量选品30天-account_name修复

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -304,11 +304,11 @@ class KafkaFlowAsinDetail(Templates):
         # 1. 关联全局缓存的df_asin_seller，用于填充seller_id为空的情况
         df = df.join(self.df_asin_seller, on=['asin'], how='left')
 
-        # 2. 优先使用kafka消息中的seller_id，为空则使用df_asin_seller的数据
+        # 2. 优先使用kafka消息数据，seller_id为空时seller_id和account_name整体取df_asin_seller数据
         df = df.withColumn(
-            "seller_id", F.coalesce(F.col("seller_id"), F.col("fd_seller_id"))
+            "account_name", F.when(F.col("seller_id").isNull(), F.col("fd_account_name")).otherwise(F.col("account_name"))
         ).withColumn(
-            "account_name", F.coalesce(F.col("account_name"), F.col("fd_account_name"))
+            "seller_id", F.when(F.col("seller_id").isNull(), F.col("fd_seller_id")).otherwise(F.col("seller_id"))
         ).drop("fd_seller_id", "fd_account_name")
 
         # 3. 关联全局缓存的df_seller_country获取seller_country_name