topic Re: Data loss with spark streaming and kafka in Data Engineering

Data loss with spark streaming and kafka

joao_augusto — Fri, 04 Apr 2025 14:04:22 GMT

Hi guys!

I'm facing a problem, and I have no idea where it came from. My process is not appending all the topic data into my bronze table. I checked the topic, and the data is there.

For example, I have some rows that are still in my Kafka topic but do not show up in my Bronze table. These records were created in Kafka around the same time yesterday. However, until now, they have not been processed, but the new data (from today) is there.

There were no errors or alerts in my jobs, and it's pure chaos.

read_kafka = (spark.readStream.format('kafka') .option("kafka.bootstrap.servers", KAFKA_HOST) .option("subscribe", topic_name) .option("startingOffsets", "earliest") .option("minPartitions", 15) .option("unparsedDataColumn", "bi_unparsed_data") .load()) read_kafka = transform_data(read_kafka, tableObject.schema, force_reset_schema) read_kafka = adjust_date_columns(read_kafka) write_stream = (read_kafka.writeStream .format("delta") .option("mergeSchema", "true") .option("checkpointLocation", f"{delta_location}_checkpoint/") .queryName(table_name_concat_target) .outputMode("Append") .trigger(once=True) .start(delta_location)) await_stream_initialization(write_stream)

Re: Data loss with spark streaming and kafka

joao_augusto — Fri, 04 Apr 2025 15:37:17 GMT

EDIT:

I checked that the data that is inserted into the kafka topic while my stream is running is neither process by the current stream nor the next job run

Re: Data loss with spark streaming and kafka

SriramMohanty — Mon, 07 Apr 2025 10:42:04 GMT

1)To Troubleshoot: Add metrics collection to your job to track the number of records processed vs. dropped.You can check the kafka offset.Compare Kafka offsets with checkpoint committed offsets.

2) Consider modifying your code to use continuous processing with a reasonable trigger interval instead of trigger(once=True) for more consistent results.

3)Temporary solution: Rename or delete the checkpoint directory to force reprocessing of all records from the beginning.