流量选品-没有变体的asin_color从product_json中提取填充

Pyspark_job/dim/dim_asin_detail.py

@@ -158,7 +158,8 @@ class DimAsinDetail(object):
             image_view as asin_image_view, spider_int as asin_spider_num, buy_sales, lob_asin_json as asin_lob_info, 
             REGEXP_REPLACE(seller_json, chr(10), '') as seller_json, buy_box_seller_type as asin_buy_box_seller_type, 
             customer_reviews_json, parent_asin, img_list, created_at as created_time, updated_at as updated_time, 
-            updated_at as dt, variat_num as variation_num, fbm_delivery_price as asin_fbm_price 
+            updated_at as dt, variat_num as variation_num, fbm_delivery_price as asin_fbm_price, 
+            get_json_object(product_json, '$.Color') as product_json_color
             from ods_asin_detail where site_name='{self.site_name}' {self.date_sql}"""
         print(sql)
         self.df_asin_detail = self.spark.sql(sqlQuery=sql)
@@ -280,7 +281,9 @@ class DimAsinDetail(object):
             )
             self.df_asin_detail = self.df_asin_detail.join(
                 df_asin_with_variation, on=['asin'], how='left'
-            )
+            ).withColumn(
+                "asin_color", F.coalesce(F.col("asin_color"), F.col("product_json_color"))
+            ).drop("product_json_color")
         elif self.date_type in ['month', 'month_week', 'month_aba_me'] and self.date_info < '2024-06':
             print("执行历史数据的变体信息整合")
             window = Window.partitionBy(self.df_asin_variat.asin).orderBy(