[Python] Spark Kafka Streaming Not Picking Up Next group_id Messages, But Batch Read Succeeds

Issue: Kafka Structured Streaming Not Processing Next group_id Messages I'm working on a PySpark Structured Streaming job that reads from a Kafka topic and processes records in batches using foreachBatch. I'm using a custom logic that filters records by a group_id field and processes one group at a time. The group_id is stored and updated in an S3 JSON file so that each batch processes only the next group.

My Setup: Spark: PySpark Structured Streaming

Kafka: Local broker (working fine)

Source: Kafka topic with messages that have a group_id column

Sink: Output written to S3 in Parquet format

Streaming Trigger: .trigger(processingTime="10 seconds")

Starting offsets: "earliest"

group_id logic: Maintained externally via a JSON file in S3

What Works: When I restart the full notebook for the first time, Spark processes messages with group_id = 1 correctly, updates the S3 group_id, and writes the predictions to S3.

The same filtering and logic work fine in batch mode (i.e., using .read() + .show()).

The Kafka topic does contain messages with group_id = 2 and beyond, and I can verify them with Kafka console consumers or through a batch query.

What Doesn’t Work: On subsequent batches, Structured Streaming does not pick up the next group_id even though the messages exist in the Kafka topic.

It seems like Spark is waiting idly and not fetching new matching records (i.e., group_id = 2).

No errors or exceptions are thrown. The stream just sits idle, and no output is written after the first group.

What I've Tried: .option("startingOffsets", "earliest") is set correctly in my Kafka readStream.

I've verified that the group_id logic only updates after data is successfully processed (i.e., rdd.isEmpty() guard is in place).

The Kafka topic has data with group_id = 2, which I can confirm using kafka-console-consumer and .read() in batch mode.

I also checked the streaming query status using query.lastProgress, and it shows no records processed after the first batch.

this is just a block of code:

kafka_df = (
spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers" , "kafka-1:29092")
.option("subscribe" , "render02")
.option("startingOffsets" , "earliest")
.load()
)
def batch_transformation(df , batch_id):

obj = s3.get_object(Bucket = "modelupload0101" , Key ="groups/current_group.json" )
group_id = json.loads(obj["Body"].read())
current_groupId = group_id["current_groupId"]
next_id = current_groupId

label = ['NORM', 'MI', 'STTC', 'CD', 'HYP']

df = df.filter(col("group_id") == next_id)

if df.rdd.isEmpty():
print(f"No data for group_id = {next_id}")
return

group_id["current_groupId"] = next_id +1

s3.put_object(Bucket = "modelupload0101" , Key = "groups/current_group.json" , Body = json.dumps(group_id) , ContentType = "application/json")


wavelength_data = df.select(col("wavelength")).rdd.map(lambda row: row["wavelength"]).collect()
y_data = np.vstack(wavelength_data)

dem_df = 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_ = row[0]
x = []

for i in x_:
x.append(i)

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 in pred:
for column in range(len(row)):
if row[column]>0.5:
preds.append(label[column])
flag = 1

if flag == 0:
preds.append(label[0])


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


(df
.write
.mode("append")
.parquet("s3a://modelupload0101/output/")
)

# batch_transformation(main_df , 2)

(
main_df
.writeStream
.foreachBatch(batch_transformation)
.trigger(processingTime = "7 seconds")
.option("checkpointLocation" , "checkpoint_kafka_dir")
.start()
.awaitTermination()
)

```


Question: Why is Structured Streaming in Spark not picking up the next group_id messages from Kafka even though the messages exist and are readable in batch mode? Is there something I’m missing about offset handling or micro-batch triggering for filtered streams?

Continue reading...