流量选品月/周-新增asin属性列表解析

Pyspark_job/dim/dim_asin_detail.py

@@ -166,7 +166,7 @@ class DimAsinDetail(object):
             get_json_object(product_json, '$.Number of Items') as product_json_number_of_items, 
             current_asin,
             nullif(coalesce(get_json_object(amazon_label, '$[0].badge_type'), get_json_object(amazon_label, '$.badge_type')), 'unknown') as amazon_label,
-            landing_asin
+            landing_asin, voyager_detail_json
             from ods_asin_detail where site_name='{self.site_name}' {self.date_sql}"""
         print(sql)
         self.df_asin_detail = self.spark.sql(sqlQuery=sql)
@@ -364,6 +364,45 @@ class DimAsinDetail(object):
         self.df_asin_detail = self.df_asin_detail.join(df_volume, on=['asin'], how='left').drop("asin_volume_type")
         self.df_asin_weight_and_volume.unpersist()
 
+    # voyager_detail_json 按小写key取值，多个key命中时值去重(忽略大小写，保留首次出现的写法)后 ' | ' 拼接
+    @staticmethod
+    def get_voyager_attr(keys):
+        keys_sql = ", ".join([f"'{k}'" for k in keys])
+        return F.expr(f"""nullif(concat_ws(' | ', aggregate(
+            transform(filter(map_entries(voyager_map), e -> lower(trim(e.key)) in ({keys_sql}) and trim(e.value) != ''),
+                      e -> trim(e.value)),
+            cast(array() as array<string>),
+            (acc, x) -> if(array_contains(transform(acc, y -> lower(y)), lower(x)), acc, concat(acc, array(x))))), '')""")
+
+    # 解析voyager_detail_json属性信息
+    def handle_asin_voyager_attribute(self):
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "voyager_map", F.from_json(F.col("voyager_detail_json"), MapType(StringType(), StringType()))
+        ).withColumn(
+            "asin_age_range", self.get_voyager_attr(["age range description"])
+        ).withColumn(
+            "asin_seasons", self.get_voyager_attr(["seasons"])
+        ).withColumn(
+            "asin_additional_features", self.get_voyager_attr(["additional features", "other special features of the product"])
+        ).withColumn(
+            "asin_closure_type", self.get_voyager_attr(["closure type", "mounting type", "installation type"])
+        ).withColumn(
+            "asin_shape", self.get_voyager_attr(["shape"])
+        ).withColumn(
+            "asin_theme", self.get_voyager_attr(["theme"])
+        ).withColumn(
+            "asin_pattern", self.get_voyager_attr(["pattern"])
+        ).withColumn(
+            "voyager_number_of_items", self.get_voyager_attr(["number of items"]).cast("int")
+        ).withColumn(
+            "voyager_number_of_pieces", self.get_voyager_attr(["number of pieces"]).cast("int")
+        ).withColumn(
+            "voyager_number_of_items", F.coalesce(
+                F.when(F.col("voyager_number_of_items") > 0, F.col("voyager_number_of_items")),
+                F.when(F.col("voyager_number_of_pieces") > 0, F.col("voyager_number_of_pieces"))
+            )
+        ).drop("voyager_map", "voyager_detail_json", "voyager_number_of_pieces")
+
     # 打包数量解析
     def get_package_quantity(self):
         self.df_asin_detail = self.df_asin_detail.repartition(100)
@@ -379,10 +418,13 @@ class DimAsinDetail(object):
         ).withColumn(
             "variat_package_quantity_is_abnormal", self.df_asin_detail.variat_parse.getField("is_package_quantity_abnormal")
         ).drop("title_parse", "variat_parse", "variat_attribute")
-        # Number of Items：直接从 product_json 提取，cast 失败（脏数据）自动为 null
+        # Number of Items：voyager(Number of Items > Number of Pieces) 优先，其次 product_json，cast 失败（脏数据）自动为 null
         self.df_asin_detail = self.df_asin_detail.withColumn(
-            "number_of_items", F.col("product_json_number_of_items").cast("int")
-        ).drop("product_json_number_of_items")
+            "number_of_items", F.coalesce(
+                F.col("voyager_number_of_items"),
+                F.when(F.col("product_json_number_of_items").cast("int") > 0, F.col("product_json_number_of_items").cast("int"))
+            )
+        ).drop("product_json_number_of_items", "voyager_number_of_items")
         # 优先级：Number of Items > 属性字段 > 标题解析 > 默认1
         self.df_asin_detail = self.df_asin_detail.withColumn(
             "package_quantity", F.expr("""CASE
@@ -540,7 +582,7 @@ class DimAsinDetail(object):
                     self.handle_string_num_value('asin_weight_type').alias('asin_weight_type'),
                     self.handle_string_num_value('asin_weight_str').alias('asin_weight_str'),
                     F.lit(None).alias('asin_package_quantity'),
-                    F.lit(None).alias('asin_pattern_name'),
+                    "asin_pattern",
                     "category_id", "category_first_id",
                     F.lit(None).alias("buy_data_bought_month"),
                     F.lit(None).alias("buy_data_bought_week"),
@@ -553,6 +595,8 @@ class DimAsinDetail(object):
                     "customer_reviews_json", "img_list", "variat_list",
                     F.round("asin_fbm_price", 2).alias("asin_fbm_price"),
                     "current_asin", "amazon_label", "keepa_tracking_since", "landing_asin",
+                    "asin_age_range", "asin_seasons", "asin_additional_features", "asin_closure_type",
+                    "asin_shape", "asin_theme",
                     F.lit(self.site_name).alias('site_name'),
                     F.lit(self.date_type).alias('date_type'),
                     F.lit(self.date_info).alias('date_info')).persist(StorageLevel.MEMORY_ONLY)
@@ -582,6 +626,8 @@ class DimAsinDetail(object):
         self.handle_asin_buy_box_seller_type()
         # 处理asin体积重量信息
         self.handle_asin_basic_attribute()
+        # 解析voyager_detail_json属性信息
+        self.handle_asin_voyager_attribute()
         # 打包数量解析
         self.get_package_quantity()
         # 获取node_id对应的分类信息(头部分类)

Pyspark_job/dwt/dwt_flow_asin.py

@@ -214,7 +214,8 @@ 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, current_asin
+            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, variat_list, current_asin,
+            asin_pattern, asin_age_range, asin_seasons, asin_additional_features, asin_closure_type, asin_shape, asin_theme, node_id
             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)
@@ -856,7 +857,7 @@ class DwtFlowAsin(Templates):
                    F.lit(None).alias("asin_freight_air_fee"), F.lit(None).alias("asin_freight_ocean_fee"),
                    F.lit(None).alias("asin_operate_fee"), F.lit(None).alias("asin_air_freight_gross_margin"),
                    F.lit(None).alias("asin_ocean_freight_gross_margin"), "asin_crawl_date",
-                   F.lit(None).alias("asin_package_quantity"), F.lit(None).alias("asin_pattern_name"),
+                   F.lit(None).alias("asin_package_quantity"), "asin_pattern",
                    "category_first_id", "category_id", "desc_category_first_id",
                    "first_category_rank", "current_category_rank",
                    "asin_weight_ratio", "asin_bought_month", F.lit(None).alias("buy_data_bought_week"),
@@ -869,6 +870,8 @@ class DwtFlowAsin(Templates):
                    "multi_color_flag", "multi_color_str", "amazon_label", "asin_weight_str", "best_sellers_herf", "best_sellers_rank",
                    "variation_bought_month", "variation_bought_month_mom", "variation_bought_month_yoy",
                    "auctions_num", "auctions_num_all", "skus_num_creat", "skus_num_creat_all",
+                   "asin_age_range", "asin_seasons", "asin_additional_features", "asin_closure_type",
+                   "asin_shape", "asin_theme", "node_id",
                    F.lit(self.site_name).alias("site_name"), F.lit(self.date_type).alias("date_type"),
                    F.lit(self.date_info).alias("date_info"))
         self.df_save = self.df_save.na.fill(

Pyspark_job/dwt/dwt_flow_asin_week.py

@@ -167,7 +167,8 @@ class DwtFlowAsinWeek(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, current_asin
+            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, current_asin,
+            asin_age_range, asin_seasons, asin_additional_features, asin_closure_type, asin_shape, asin_theme, asin_pattern, node_id
             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)
@@ -866,6 +867,8 @@ class DwtFlowAsinWeek(Templates):
             F.col("asin_bought_wow_last_3w").cast("double").alias("asin_bought_wow_last_3w"),
             F.col("asin_bought_rise").cast("int").alias("asin_bought_rise"),
             F.col("variation_bought_month_rise").cast("int").alias("variation_bought_month_rise"),
+            "asin_age_range", "asin_seasons", "asin_additional_features", "asin_closure_type",
+            "asin_shape", "asin_theme", "asin_pattern", "node_id",
             F.lit(self.site_name).alias("site_name"),
             F.lit(self.date_type).alias("date_type"),
             F.lit(self.date_info).alias("date_info")