流量选品30天-兼容信息库流程

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -115,7 +115,10 @@ class KafkaFlowAsinDetail(Templates):
         self.df_self_asin = self.spark.sql("select 1+1;")
         self.df_self_brand = self.spark.sql("select 1+1;")
         self.df_hide_category = self.spark.sql("select 1+1;")
+        self.df_ai_hide_category = self.spark.sql("select 1+1;")
         self.color_set = set()
+        self.save_parent_asin_latest_detail_cache = None
+        self.save_asin_latest_detail_cache = None
         # udf函数注册
         package_schema = StructType([
             StructField("parse_package_quantity", IntegerType(), True),
@@ -205,6 +208,8 @@ class KafkaFlowAsinDetail(Templates):
             StructField("follow_sellers", IntegerType(), True),
             StructField("fbm_delivery_price", FloatType(), True),
             StructField("product_json", StringType(), True),
+            StructField("product_detail_json", StringType(), True),
+            StructField("review_json_list", StringType(), True),
             StructField("amazon_label", StringType(), True),
             StructField("sub_title", StringType(), True)
         ])
@@ -460,7 +465,7 @@ class KafkaFlowAsinDetail(Templates):
         df = df.withColumn(
             "number_of_items",
             F.get_json_object(F.col("product_json"), "$.Number of Items").cast("int")
-        ).drop("product_json")
+        )  # product_json 信息库PG导出需要用到，不再drop
         # 优先级：Number of Items > 属性字段 > 标题解析 > 默认1
         df = df.withColumn(
             "package_quantity", F.expr("""
@@ -1001,6 +1006,21 @@ class KafkaFlowAsinDetail(Templates):
             username=mysql_con['username'], query=sql
         ).persist(StorageLevel.MEMORY_ONLY))
         self.df_hide_category.show(10, truncate=False)
+        print("11. 读取禁用分类(信息库模块，id_path前缀匹配)，用于信息库PG导出过滤")
+        # 跟df_hide_category同样的匹配方式，只是module换成信息库自己的配置，避免误用流量选品的禁用分类
+        sql = f"""
+            SELECT DISTINCT category_id FROM category_full_name a
+            WHERE EXISTS (
+                SELECT 1 FROM category_disable_config b
+                WHERE b.site = a.site AND b.module = '信息库:名称前缀筛选'
+                  AND a.id_path LIKE CONCAT(b.id_path, '%')
+            ) AND a.site = '{self.site_name}'
+        """
+        self.df_ai_hide_category = F.broadcast(SparkUtil.read_jdbc_query(
+            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+            username=mysql_con['username'], query=sql
+        ).persist(StorageLevel.MEMORY_ONLY))
+        self.df_ai_hide_category.show(10, truncate=False)
 
     # 字段处理逻辑综合
     def handle_all_field(self, df):
@@ -1018,6 +1038,10 @@ class KafkaFlowAsinDetail(Templates):
         df = self.handle_asin_basic_attribute_info(df)
         # 7. 处理asin图片信息
         df = self.handle_asin_img_info(df)
+        # persist：第8步内部(test_flag=normal时会写一次parent_asin_latest_detail)会触发一次action，
+        # 不persist的话后面第8~14步还会把第1~7步重新算一遍。批次末尾统一unpersist，见_unpersist_batch_cache
+        df = df.persist(StorageLevel.DISK_ONLY)
+        self.save_parent_asin_latest_detail_cache = df
         # 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
         df = self.handle_asin_measure(df)
         # 9. 提取打包数量字段
@@ -1030,12 +1054,27 @@ class KafkaFlowAsinDetail(Templates):
         df = self.handle_asin_detail_all_type(df)
         # 12. 处理变化率相关字段
         df = self.handle_asin_attribute_change(df)
+        # persist：第13步(asin_latest_detail写入)、信息库PG导出、字段标准化后的doris主表写入，
+        # 三个action共用这份结果，避免第8~12步被重复算3次
+        df = df.persist(StorageLevel.DISK_ONLY)
+        self.save_asin_latest_detail_cache = df
         # 13. 写入 ASIN 最新详情表（在字段标准化 select 过滤前，趁 category/seller_json 等原始字段还在）
         self.save_asin_latest_detail(df)
+        # 13.5 导出信息库ASIN详情到PG（倒数第二步，同样要在字段标准化前，用的还是原始字段名）
+        self.save_ai_asin_detail(df)
         # 14. 字段标准化
         df_save = self.handle_column_name(df)
         return df_save
 
+    def _unpersist_batch_cache(self):
+        """及时释放handle_all_field里persist的中间结果，避免多批次/多次重试导致缓存堆积"""
+        if self.save_parent_asin_latest_detail_cache is not None:
+            self.save_parent_asin_latest_detail_cache.unpersist()
+            self.save_parent_asin_latest_detail_cache = None
+        if self.save_asin_latest_detail_cache is not None:
+            self.save_asin_latest_detail_cache.unpersist()
+            self.save_asin_latest_detail_cache = None
+
     @staticmethod
     def udf_rank_and_category(best_sellers_rank, pattern_str, top100_prefix):
         import re
@@ -1150,6 +1189,68 @@ class KafkaFlowAsinDetail(Templates):
         end_time = time.time()
         print(f"Doris {self.doris_30day_table} 写入完毕，耗时：{end_time - start_time:.1f}s")
 
+    # 导出信息库ASIN详情到PG（仅normal+latest模式；允许重复写入，同事AI分析工具按asin去重读取，不做唯一性判重）
+    # 注意：在handle_column_name(字段标准化)之前调用，字段名都是renamed前的原始名
+    def save_ai_asin_detail(self, df):
+        if self.test_flag != 'normal' or self.consumer_type != 'latest':
+            return
+        print("导出信息库ASIN详情到PG：")
+        start_time = time.time()
+        df_info_base = df.filter(
+            "asin_type in (0, 1) and asin_bought_month >= 50"
+        ).join(
+            self.df_ai_hide_category,
+            df["asin_bs_cate_current_id"] == self.df_ai_hide_category["category_id"],
+            'left_anti'
+        )
+        df_info_base = df_info_base.select(
+            F.lit(self.site_name).alias('site_name'),
+            "asin",
+            F.round(F.col('weight').cast('double'), 2).alias('weight'),
+            F.col('asin_bought_month').alias('bought_month'),
+            "category",
+            F.col('img_url').alias('img'),
+            "title",
+            "brand",
+            "account_name",
+            "buy_box_seller_type",
+            F.date_format(F.to_timestamp(F.col('launch_time')), 'yyyy-MM-dd').alias('launch_time'),
+            "img_num",
+            F.when(F.col('variat_num') > 0, F.lit(1)).otherwise(F.lit(0)).alias('variation_flag'),
+            F.col('variat_num').alias('variation_num'),
+            F.round(F.col('ao_val').cast('double'), 2).alias('ao_val'),
+            F.col('asin_bs_cate_1_id').alias('category_first_id'),
+            F.col('asin_bs_cate_current_id').alias('category_current_id'),
+            "parent_asin",
+            F.col('asin_bs_cate_1_rank').alias('bsr_rank'),
+            F.round(F.col('price').cast('double'), 2).alias('price'),
+            F.round(F.col('rating').cast('double'), 2).alias('rating'),
+            "total_comments",
+            "seller_id",  # 已在第5步 coalesce 过（优先kafka自带，缺失时补df_asin_seller），跟新表seller_id同名
+            F.col('seller_country_name').alias('fb_country_name'),
+            "review_json_list",
+            "describe",
+            "product_json",
+            "product_detail_json",
+            "bought_month_mom",
+            "bought_month_yoy",
+            F.lit(0).cast('short').alias('chat_flag'),
+        ).na.fill({'total_comments': 0, 'img_num': 0}).persist(StorageLevel.DISK_ONLY)
+        export_count = df_info_base.count()
+        print(f"待导出信息库PG的ASIN数量：{export_count}")
+        if export_count > 0:
+            pg_con = DBUtil.get_connection_info("postgresql", "us")
+            df_info_base.write.format("jdbc") \
+                .option("url", pg_con["url"]) \
+                .option("dbtable", f"{self.site_name}_ai_asin_detail") \
+                .option("user", pg_con["username"]) \
+                .option("password", pg_con["pwd"]) \
+                .mode("append") \
+                .save()
+        df_info_base.unpersist()
+        end_time = time.time()
+        print(f"信息库PG导出完毕，耗时：{end_time - start_time:.1f}s")
+
     # 实时消费中批次数据的处理逻辑（latest 模式）：失败重试2次，仍失败则抛出异常（避免checkpoint误提交导致丢数据）
     def handle_kafka_stream(self, df, batch_id):
         max_retries = 3
@@ -1164,7 +1265,6 @@ class KafkaFlowAsinDetail(Templates):
                 df_repartitioned = df.repartition(self.repartition_num)
                 df_save = self.handle_all_field(df_repartitioned)
                 self.save_to_doris(df_save, batch_num)
-                df_save.unpersist()
                 end_time = time.time()
                 print(f"当前批次:{batch_id} 执行完毕, 执行时长为：{end_time - start_time:.1f}s")
                 return
@@ -1176,18 +1276,24 @@ class KafkaFlowAsinDetail(Templates):
                     )
                     raise
                 time.sleep(10)
+            finally:
+                # 及时释放本批次persist的缓存（成功/失败/每次重试都要释放），避免资源堆积
+                self._unpersist_batch_cache()
 
     # 消费主题下的所有历史数据
     def handle_kafka_history(self, kafka_df):
         print("处理kafka历史数据")
         batch_num = kafka_df.count()
         if batch_num > 0:
+            try:
                 start_time = time.time()
                 kafka_df = kafka_df.repartition(self.repartition_num)
                 kafka_df = self.handle_all_field(kafka_df)
                 self.save_to_doris(kafka_df, batch_num)
                 end_time = time.time()
                 print("该批次数据处理完毕, 执行时长为：" + str(end_time - start_time))
+            finally:
+                self._unpersist_batch_cache()
         else:
             raise ValueError("当前主题中没有数据,请注意检查！")