流量选品30天-增加asin属性列表解析

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -212,7 +212,8 @@ class KafkaFlowAsinDetail(Templates):
             StructField("product_detail_json", StringType(), True),
             StructField("review_json_list", StringType(), True),
             StructField("amazon_label", StringType(), True),
-            StructField("sub_title", StringType(), True)
+            StructField("sub_title", StringType(), True),
+            StructField("voyager_detail_json", StringType(), True)
         ])
         return schema
 
@@ -265,7 +266,7 @@ class KafkaFlowAsinDetail(Templates):
         df = df.join(self.df_asin_new_cate, on=['node_id'], how='left')
         df = df.withColumn("asin_bs_cate_current_id", F.coalesce(F.col("asin_bs_cate_current_id"), F.col("node_id"))). \
             withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("category_first_id"))). \
-            drop("category_first_id", "node_id")
+            drop("category_first_id")
         df = df.withColumn("desc_category_first_name", F.lower(F.trim(F.split(F.col("category"), "›").getItem(0))))
         df = df.join(self.df_asin_category, on=['desc_category_first_name'], how='left')
         df = df.withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("desc_category_first_id"))). \
@@ -445,6 +446,46 @@ class KafkaFlowAsinDetail(Templates):
             DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db_selection, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
         return df
 
+    # 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))))), '')""")
+
+    # 8.5 解析voyager_detail_json属性信息
+    def handle_asin_voyager_attribute(self, df):
+        df = df.withColumn(
+            "voyager_map", F.from_json(F.col("voyager_detail_json"), MapType(StringType(), StringType()))
+        ).withColumn(
+            "age_range", self.get_voyager_attr(["age range description"])
+        ).withColumn(
+            "seasons", self.get_voyager_attr(["seasons"])
+        ).withColumn(
+            "additional_features", self.get_voyager_attr(["additional features", "other special features of the product"])
+        ).withColumn(
+            "closure_type", self.get_voyager_attr(["closure type", "mounting type", "installation type"])
+        ).withColumn(
+            "shape", self.get_voyager_attr(["shape"])
+        ).withColumn(
+            "theme", self.get_voyager_attr(["theme"])
+        ).withColumn(
+            "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")
+        return df
+
     # 9. 提取打包数量字段
     def handle_asin_package_quantity(self, df):
         df = df.withColumn("material",
@@ -464,11 +505,14 @@ class KafkaFlowAsinDetail(Templates):
             "_product_json_color",
             F.lower(F.get_json_object(F.col("product_json"), "$.Color"))
         )
-        # Number of Items：从 product_json 提取，cast 失败（脏数据）自动为 null，提取后立即 drop
+        # Number of Items：voyager(Number of Items > Number of Pieces) 优先，其次 product_json，cast 失败（脏数据）自动为 null
         df = df.withColumn(
-            "number_of_items",
-            F.get_json_object(F.col("product_json"), "$.Number of Items").cast("int")
-        )  # product_json 信息库PG导出需要用到，不再drop
+            "number_of_items", F.coalesce(
+                F.col("voyager_number_of_items"),
+                F.when(F.get_json_object(F.col("product_json"), "$.Number of Items").cast("int") > 0,
+                       F.get_json_object(F.col("product_json"), "$.Number of Items").cast("int"))
+            )
+        ).drop("voyager_number_of_items")  # product_json 信息库PG导出需要用到，不再drop
         # 优先级：Number of Items > 属性字段 > 标题解析 > 默认1
         df = df.withColumn(
             "package_quantity", F.expr("""
@@ -859,6 +903,10 @@ class KafkaFlowAsinDetail(Templates):
             # ── 竞卖/SKU创建 ───────────────────────────────────────────────
             "auctions_num", "auctions_num_all",  # get_asin_variant_attribute，ASIN竞卖数/父asin累计竞卖数
             "skus_num_creat", "skus_num_creat_all",  # get_asin_variant_attribute，ASIN创建SKU数/父asin累计创建SKU数
+            # ── voyager属性/最小分类 ─────────────────────────────────────
+            "age_range", "seasons", "additional_features", "closure_type",  # voyager_detail_json 解析
+            "shape", "theme", "pattern",                                    # voyager_detail_json 解析
+            "node_id",                                                      # Kafka，最小分类 ID
         )
         df_save = df_save.na.fill(
             {"zr_counts": 0, "sp_counts": 0, "sb_counts": 0, "vi_counts": 0, "bs_counts": 0, "ac_counts": 0,
@@ -1075,6 +1123,8 @@ class KafkaFlowAsinDetail(Templates):
         self.save_parent_asin_latest_detail_cache = df
         # 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
         df = self.handle_asin_measure(df)
+        # 8.5 解析voyager_detail_json属性信息
+        df = self.handle_asin_voyager_attribute(df)
         # 9. 提取打包数量字段
         df = self.handle_asin_package_quantity(df)
         # 9.5. 多色判断（依赖 color 字段已从变体属性+product_json 补全）
@@ -1214,7 +1264,8 @@ class KafkaFlowAsinDetail(Templates):
             is_contains_lob_info, is_package_quantity_abnormal, zr_flow_proportion, matrix_flow_proportion,
             matrix_ao_val, product_features, img_info, collapse_asin, follow_sellers_count, asin_describe,
             fbm_price, describe_len, multi_color_flag, multi_color_str, amazon_label, is_amazon_new,
-            auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all"""
+            auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all,
+            age_range, seasons, additional_features, closure_type, shape, theme, pattern, node_id"""
         print(f"写入Doris {self.doris_30day_table}")
         DorisHelper.spark_export_with_columns(
             df_save=df, db_name=self.doris_db_dwt,