keepa数据增加类型解析

Pyspark_job/dim/dim_keepa_asin_info.py

@@ -63,7 +63,9 @@ class DimKeepaAsinInfo(object):
             F.get_json_object("last_detail", "$.trackingSince").cast("int").alias("tracking_since"),
             F.get_json_object("last_detail", "$.productType").cast("int").alias("product_type"),
             F.get_json_object("last_detail", "$.title").alias("title"),
-            F.col('updated_time')
+            F.col('updated_time'),
+            F.lower(F.regexp_replace(F.get_json_object("last_detail", "$.type"), "_", "-")).alias("keepa_type"),
+            F.lower(F.regexp_replace(F.get_json_object("last_detail", "$.itemTypeKeyword"), "_", "-")).alias("keepa_item_type")
         ).withColumn(
             # weight 字段已废弃（Doris 端 dwd_keepa_asin_detail 不再保留），Hive 兼容 schema 保留列但置 NULL
             'weight', F.lit(None).cast('int')
@@ -93,7 +95,9 @@ class DimKeepaAsinInfo(object):
             F.col('release_date'),
             F.col('tracking_since'),
             F.col('product_type'),
-            F.col('title')
+            F.col('title'),
+            F.col('keepa_type'),
+            F.col('keepa_item_type')
         ).filter(
             # 过滤脏数据：productType in (3,4,5) 且 title 为空的异常数据不做保留
             ~(F.col("product_type").isin(3, 4, 5) & F.col("title").isNull())
@@ -105,12 +109,13 @@ class DimKeepaAsinInfo(object):
             'asin', 'package_length', 'package_width', 'package_height', 'package_weight', 'item_weight',
             'listed_since', 'release_date', 'tracking_since',
             F.to_timestamp(F.col('keepa_launch_time')).alias('keepa_launch_time'),
-            F.to_timestamp(F.col('updated_time')).alias('updated_time')
+            F.to_timestamp(F.col('updated_time')).alias('updated_time'),
+            'keepa_type', 'keepa_item_type'
         )
 
         # 读取历史数据
         sql = f"""
-            select asin, package_length, package_width, package_height, package_weight, item_weight, weight, keepa_launch_time, updated_time, listed_since, release_date, tracking_since, product_type, title
+            select asin, package_length, package_width, package_height, package_weight, item_weight, weight, keepa_launch_time, updated_time, listed_since, release_date, tracking_since, product_type, title, keepa_type, keepa_item_type
             from dim_keepa_asin_info where site_name = '{self.site_name}';
         """
         self.df_keepa_asin_history = self.spark.sql(sqlQuery=sql).repartition(40, 'asin').cache()
@@ -167,7 +172,7 @@ class DimKeepaAsinInfo(object):
         print("往doris存储最新keepa详情信息：")
         doris_table_columns = """
         site_name, asin, package_length, package_width, package_height, package_weight, item_weight,
-        listed_since, release_date, tracking_since, keepa_launch_time, updated_time
+        listed_since, release_date, tracking_since, keepa_launch_time, updated_time, keepa_type, keepa_item_type
         """
         DorisHelper.spark_export_with_columns(
             df_save=self.df_to_doris,