kafka实时消费-latest模式下offset偏差过大特殊情况处理

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -29,7 +29,7 @@ NEED_FILTER_CATEGORIES = (
 
 
 class KafkaFlowAsinDetail(Templates):
-    def __init__(self, site_name='us', date_type="month", date_info='2026-03', consumer_type='history', test_flag='test', batch_size=100000):
+    def __init__(self, site_name='us', date_type="month", date_info='2026-03', consumer_type='history', test_flag='test', batch_size=20000):
         super().__init__()
         self.site_name = site_name
         self.date_type = date_type
@@ -1308,5 +1308,5 @@ if __name__ == '__main__':
         test_flag = sys.argv[5]
     else:
         test_flag = 'normal'
-    handle_obj = KafkaFlowAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info, consumer_type=consumer_type, test_flag=test_flag, batch_size=200000)
+    handle_obj = KafkaFlowAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info, consumer_type=consumer_type, test_flag=test_flag, batch_size=20000)
     handle_obj.run_kafka()

Pyspark_job/utils/templates.py

@@ -120,7 +120,7 @@ class Templates(object):
     def create_kafka_df_object(self, consumer_type=str(), topic_name=str(), starting_offsets_json=str(), ending_offsets_json=str(), schema=StructType()):
         if consumer_type == "latest":
             # 流处理
-            kafka_df = self.spark.readStream \
+            kafka_stream_reader = self.spark.readStream \
                     .format("kafka") \
                     .option("kafka.bootstrap.servers", self.kafka_servers) \
                     .option("subscribe", topic_name) \
@@ -129,8 +129,12 @@ class Templates(object):
                     .option("kafka.sasl.jaas.config",
                             f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{self.kafka_username}" password="{self.kafka_password}";') \
                     .option("startingOffsets", consumer_type) \
-                    .option("failOnDataLoss", "false") \
-                    .load() \
+                    .option("failOnDataLoss", "false")
+            # 断点与topic最新offset差距过大时，避免单个触发批次读取过多数据导致阻塞/失败风险，按批次上限分批追平
+            max_offsets_per_trigger = getattr(self, 'batch_size', None)
+            if max_offsets_per_trigger:
+                kafka_stream_reader = kafka_stream_reader.option("maxOffsetsPerTrigger", max_offsets_per_trigger)
+            kafka_df = kafka_stream_reader.load() \
                     .select(F.from_json(F.col("value").cast("string"), schema=schema).alias("data")) \
                     .select("data.*")
             return kafka_df