流量选品-标题解析迭代-拼接正副标题

Pyspark_job/dim/dim_asin_detail.py

@@ -145,8 +145,10 @@ class DimAsinDetail(object):
         self.df_asin_keep_date.show(10, truncate=False)
         print("2. 获取ods_asin_detail")
         sql = f"""
-            select asin, img_url as asin_img_url, lower(title) as asin_title, title_len as asin_title_len, 
-            price as asin_price, rating as asin_rating, total_comments as asin_total_comments, 
+            select asin, img_url as asin_img_url,
+            lower(concat_ws(' | ', title, nullif(trim(sub_title), ''))) as asin_title,
+            length(concat_ws(' | ', title, nullif(trim(sub_title), ''))) as asin_title_len,
+            price as asin_price, rating as asin_rating, total_comments as asin_total_comments,
             page_inventory as asin_page_inventory, category as asin_category_desc, 
             launch_time as crawl_asin_launch_time, img_num as asin_img_num, img_type as asin_img_type, 
             category_state as asin_category_state, material as asin_material, lower(brand) as asin_brand_name, 

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -205,15 +205,23 @@ class KafkaFlowAsinDetail(Templates):
             StructField("follow_sellers", IntegerType(), True),
             StructField("fbm_delivery_price", FloatType(), True),
             StructField("product_json", StringType(), True),
-            StructField("amazon_label", StringType(), True)
+            StructField("amazon_label", StringType(), True),
+            StructField("sub_title", StringType(), True)
         ])
         return schema
 
-    # 覆写模板去重方法：去重前先处理跳转asin（判断规则见 resolve_asin_jump），
-    # 确保跳转后同 asin 多条记录在去重时被合并（在去重前执行）
+    # 覆写模板去重方法：第一步处理，在去重前统一完成——
+    # 1) 处理跳转asin（判断规则见 resolve_asin_jump），确保跳转后同 asin 多条记录在去重时被合并
+    # 2) 标题拼接+转小写：title+sub_title用' | '拼接后统一转小写
     def deduplication_kafka_data(self, kafka_df, deduplicaiton_key_field, deduplication_time_field):
         if deduplicaiton_key_field == "asin" and "current_asin" in kafka_df.columns:
             kafka_df = kafka_df.withColumn("asin", resolve_asin_jump())
+        kafka_df = kafka_df.withColumn(
+            "title", F.lower(F.concat_ws(
+                " | ", F.col("title"),
+                F.when(F.trim(F.col("sub_title")) == '', F.lit(None)).otherwise(F.trim(F.col("sub_title")))
+            ))
+        ).withColumn("title_len", F.length(F.col("title"))).drop("sub_title")
         return super().deduplication_kafka_data(kafka_df, deduplicaiton_key_field, deduplication_time_field)
 
     # 1. 处理asin分类及排名以及排名类型字段
@@ -468,7 +476,6 @@ class KafkaFlowAsinDetail(Templates):
                  ELSE 2 END""")
         ).drop("number_of_items", "title_package_quantity", "variat_package_quantity",
                "title_package_quantity_is_abnormal", "variat_package_quantity_is_abnormal")
-        df = df.withColumn("title", F.lower(F.col("title")))
         # color 优先使用变体属性颜色，fallback 到 product_json 中的 Color 字段
         df = df.withColumn("color", F.coalesce(F.col("color"), F.col("_product_json_color"))).drop("_product_json_color")
         return df

Pyspark_job/sqoop_import/ods_asin_detail.py

@@ -42,7 +42,7 @@ if __name__ == '__main__':
            "sp_4stars_initial_seen_asins_json, sp_delivery_initial_seen_asins_json, compare_similar_asin_json, " \
            "together_asin_json, min_match_asin_json, variat_num, current_asin, img_list, variat_list, parent_asin, " \
            "bundles_this_asins_json, video_m3u8_url, result_list_json, bundle_asin_component_json, review_json_list, " \
-           "fbm_delivery_price, amazon_label, landing_asin"
+           "fbm_delivery_price, amazon_label, landing_asin, sub_title"
 
     engine = get_remote_engine(
         site_name=site_name,
@@ -145,7 +145,8 @@ if __name__ == '__main__':
   REPLACE(REPLACE(REPLACE(review_json_list, E'\n',' '), E'\r',' '), E'\t',' ') AS review_json_list,
   fbm_delivery_price,
   REPLACE(REPLACE(REPLACE(amazon_label, E'\n',' '), E'\r',' '), E'\t',' ') AS amazon_label,
-  REPLACE(REPLACE(REPLACE(landing_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS landing_asin
+  REPLACE(REPLACE(REPLACE(landing_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS landing_asin,
+  REPLACE(REPLACE(REPLACE(sub_title, E'\n',' '), E'\r',' '), E'\t',' ') AS sub_title
 FROM {import_table}
 WHERE  1=1 AND \$CONDITIONS
 """