Spark Kafka结构化流无法读取下一个group_id消息，批处理正常

你的核心问题是Spark结构化流的偏移量提交机制与分组处理逻辑不匹配：

1. 首次启动时，Spark会从earliest偏移量拉取所有可用消息（包括group_id=1和group_id=2的消息）到第一个微批中。
2. 在foreachBatch里你只过滤处理了group_id=1的消息，但Spark会默认将整个微批的偏移量标记为已处理并写入checkpoint。
3. 后续批次启动时，Spark会从checkpoint记录的偏移量开始拉取新消息，而group_id=2的消息已经被包含在第一个微批的偏移量范围内，不会被再次拉取，导致后续批次无数据可处理。

而批处理模式下每次都是全量读取Kafka消息，因此能正常读到group_id=2的数据。

方法1：手动管理Kafka偏移量，禁用自动提交 #

修改配置禁用Spark自动提交偏移量，仅在处理完目标group_id的消息后，手动提交对应偏移量：

1. 调整Kafka读取配置 #

kafka_df = (
    spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-1:29092")
    .option("subscribe", "render02")
    .option("startingOffsets", "earliest")
    .option("enable.auto.commit", "false")  # 禁用自动提交
    .option("kafka.group.id", "spark-stream-group")  # 指定消费者组ID
    .load()
    # 这里补充你的JSON解析逻辑，将value字段解析为包含group_id的结构化数据
    .select(...)
)

2. 修改foreachBatch逻辑，手动提交偏移量 #

from kafka import KafkaProducer
import json
from pyspark.sql.functions import col, max, lit, array, when

def batch_transformation(df, batch_id):
    # 读取当前待处理的group_id
    obj = s3.get_object(Bucket="modelupload0101", Key="groups/current_group.json")
    group_info = json.loads(obj["Body"].read())
    current_group_id = group_info["current_groupId"]
    
    # 保留Kafka原始偏移量字段，用于后续提交
    df_with_offset = df.select(
        col("group_id"),
        col("wavelength"),
        col("infractum_stad_1"),
        col("infractum_stad_2"),
        col("pacemaker"),
        col("age"),
        col("sex"),
        col("height"),
        col("weight"),
        col("topic"),
        col("partition"),
        col("offset")
    )
    
    # 过滤目标group_id的数据
    target_df = df_with_offset.filter(col("group_id") == current_group_id)
    
    if target_df.rdd.isEmpty():
        print(f"No data for group_id = {current_group_id}")
        return
    
    # --------------------------
    # 原有数据处理逻辑
    # --------------------------
    label = ['NORM', 'MI', 'STTC', 'CD', 'HYP']
    wavelength_data = target_df.select(col("wavelength")).rdd.map(lambda row: row["wavelength"]).collect()
    y_data = np.vstack(wavelength_data)

    dem_df = target_df.select(col("infractum_stad_1"), col("infractum_stad_2"), col("pacemaker"), col("age"), col("sex"), col("height"), col("weight")).limit(1)
    row = dem_df.collect()

    if not row:
        return 
        
    x = [i for i in row[0]]
    x = np.array([x])
    y_data = np.array([y_data])
    print(x.shape, y_data.shape)
    pred = model.predict([x, y_data])

    preds = []
    flag = 0
    for row_pred in pred:
        for column in range(len(row_pred)):
            if row_pred[column]>0.5:
                preds.append(label[column])
                flag = 1
    if flag == 0:
        preds.append(label[0])

    target_df = target_df.withColumn(
        "prediction",
        when(col("group_id") == current_group_id, array([lit(p) for p in preds])).otherwise(None)
    )

    target_df.write.mode("append").parquet("s3a://modelupload0101/output/")
    
    # --------------------------
    # 手动提交当前group_id的最大偏移量
    # --------------------------
    # 按分区获取当前处理的最大偏移量
    max_offset_df = target_df.groupBy("topic", "partition").agg(max("offset").alias("max_offset"))
    max_offsets = max_offset_df.collect()
    
    # 构造Kafka偏移量提交格式
    offset_commit = {}
    for row in max_offsets:
        topic = row["topic"]
        partition = row["partition"]
        # 提交到下一个偏移量，表示已处理到当前max_offset
        offset_commit.setdefault(topic, {})[partition] = row["max_offset"] + 1
    
    # 提交偏移量到Kafka消费者组
    producer = KafkaProducer(bootstrap_servers="kafka-1:29092")
    producer.send("__consumer_offsets", json.dumps(offset_commit).encode('utf-8'))
    producer.flush()
    
    # 更新S3中的待处理group_id
    group_info["current_groupId"] = current_group_id + 1
    s3.put_object(
        Bucket="modelupload0101", 
        Key="groups/current_group.json", 
        Body=json.dumps(group_info), 
        ContentType="application/json"
    )

方法2：改用定时批处理循环模拟流处理 #

如果手动管理偏移量过于复杂，可放弃结构化流自动触发，改用定时批处理循环，每次仅处理当前group_id的消息：

import time
from pyspark.sql.functions import col, lit, array, when

while True:
    # 读取当前待处理的group_id
    obj = s3.get_object(Bucket="modelupload0101", Key="groups/current_group.json")
    group_info = json.loads(obj["Body"].read())
    current_group_id = group_info["current_groupId"]
    
    # 批处理读取Kafka并过滤目标group_id
    batch_df = (
        spark
        .read
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka-1:29092")
        .option("subscribe", "render02")
        .option("startingOffsets", "earliest")
        .load()
        # 补充JSON解析逻辑，解析出group_id等字段
        .select(...)
        .filter(col("group_id") == current_group_id)
    )
    
    if batch_df.rdd.isEmpty():
        print(f"No data for group_id = {current_group_id}, waiting...")
        time.sleep(10)
        continue
    
    # --------------------------
    # 原有数据处理逻辑（同foreachBatch内的处理）
    # --------------------------
    label = ['NORM', 'MI', 'STTC', 'CD', 'HYP']
    wavelength_data = batch_df.select(col("wavelength")).rdd.map(lambda row: row["wavelength"]).collect()
    y_data = np.vstack(wavelength_data)

    dem_df = batch_df.select(col("infractum_stad_1"), col("infractum_stad_2"), col("pacemaker"), col("age"), col("sex"), col("height"), col("weight")).limit(1)
    row = dem_df.collect()

    if not row:
        time.sleep(10)
        continue 
        
    x = [i for i in row[0]]
    x = np.array([x])
    y_data = np.array([y_data])
    print(x.shape, y_data.shape)
    pred = model.predict([x, y_data])

    preds = []
    flag = 0
    for row_pred in pred:
        for column in range(len(row_pred)):
            if row_pred[column]>0.5:
                preds.append(label[column])
                flag = 1
    if flag == 0:
        preds.append(label[0])

    batch_df = batch_df.withColumn(
        "prediction",
        when(col("group_id") == current_group_id, array([lit(p) for p in preds])).otherwise(None)
    )

    batch_df.write.mode("append").parquet("s3a://modelupload0101/output/")
    
    # 更新S3中的待处理group_id
    group_info["current_groupId"] = current_group_id + 1
    s3.put_object(
        Bucket="modelupload0101", 
        Key="groups/current_group.json", 
        Body=json.dumps(group_info), 
        ContentType="application/json"
    )
    
    time.sleep(10)

1. 方法1需确保处理逻辑是幂等的，避免重复处理数据；
2. 方法2的批处理循环更简单，但需自行维护定时逻辑，且要额外处理失败重试；
3. 无论哪种方法，都要保证S3中的current_group.json是原子更新的，可利用S3的条件写入或版本控制避免并发冲突。

内容的提问来源于stack exchange，提问作者117__pushpak raj__