spark消费停止前，把kafka已有数据全部消费完再停止

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -977,7 +977,7 @@ class KafkaFlowAsinDetail(Templates):
         self.df_self_brand = F.broadcast(SparkUtil.read_jdbc_query(
             session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
             username=mysql_con['username'], query=mysql_brand_sql
-        ))
+        ).persist(StorageLevel.MEMORY_ONLY))
         self.df_self_brand.show(10, truncate=False)
         print("10. 读取隐藏分类(流量选品模块，id_path前缀匹配)，用于asin_type计算")
         # category_full_name + category_disable_config 按 id_path 前缀匹配（对齐 dwt.handle_asin_is_hide）
@@ -992,7 +992,7 @@ class KafkaFlowAsinDetail(Templates):
         self.df_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_hide_category.show(10, truncate=False)
 
     # 字段处理逻辑综合

Pyspark_job/utils/templates.py

@@ -348,6 +348,9 @@ class Templates(object):
         def _do_stop():
             try:
                 if self.query is not None:
+                    # 先把调用时刻Kafka里已经存在但还没被消费的积压全部跑完，再停，
+                    # 避免爬虫刚标记完成、但还有一截尾巴数据没被下一次触发扫到就被stop掉
+                    self.query.processAllAvailable()
                     self.query.stop()  # 在子线程中调用，避免 foreachBatch 回调内死锁
                 if self.spark is not None:
                     self.spark.stop()