How to restart failed spark stream job from the failure point

[Rajdeepak]

I am setting up a ETL process using pyspark. My input is a kafka stream and i am writing output to multiple sink (one into kafka and another into cloud storage). I am writing checkpoints on the cloud storage. The issue i am facing is that, whenever my application is getting failed due to some reason and when i am restarting my application then, my pyspark application is again reprocessing some (not all) of the input stream data causing data redundancy. Is there any way i can avoid this. I am using spark 3.5.0 and python 3.11. Below are some of my application code:

Spark Session :

spark = SparkSession \
.builder \
.appName("ETL") \
.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.2") \
.config('spark.hadoop.fs.s3a.aws.credentials.provider', 'org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider')\
.config('spark.driver.extraJavaOptions', '-Duser.timezone=GMT') \
.config('spark.executor.extraJavaOptions', '-Duser.timezone=GMT') \
.config('spark.sql.session.timeZone', 'UTC') \
.config('spark.hadoop.fs.s3a.buffer.dir', '/tmp,/mnt/tmp') \
.config('spark.hadoop.fs.s3a.fast.upload.buffer', 'bytebuffer') \
.config('spark.hadoop.fs.s3a.fast.upload.active.blocks', 1) \
.config('spark.streaming.backpressure.enabled', True) \
.config("spark.redis.host",conf["nosql-host"]) \
.config("spark.redis.port",conf["nosql-port"]) \
.config("spark.redis.db",conf["nosql-db"]) \
.config("spark.redis.auth", __REDIS_CREDENTIAL__) \
.getOrCreate()  

Kafka Read Stream :

streamDF = (spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_bootstrap_server_consumer) \
.option("subscribe", kafka_topic_name) \
.option("mode", "PERMISSIVE") \
.option("startingOffsets", "earliest").option("failOnDataLoss", "false") \
.load().withColumn('fixedValue', fn.expr("substring(value, 6, length(value)-5)")).select('fixedValue'))

Write Stream to multiple sinks :

write_stream = extractionDF \
.writeStream \
.trigger(processingTime='2 seconds') \
.outputMode("append") \
.foreachBatch(lambda df,epochId: write_to_multiple_sinks(df, epochId,processed_cloud_storage_path,kafka_bootstrap_server_producer)) \
.option("truncate", "false").option("checkpointLocation", cloud_storage_path)\
.start()

write_to_multiple_sinks Function :

def write_to_multiple_sinks(dataframe: DataFrame, epochId,cloud_storage_path, kafka_bootstrap_server):
dataframe = dataframe.cache()
druidDF = dataframe.select(druidSchema())

druidDF.selectExpr(producerTopic,"to_json(struct(*)) AS value").write\
.format("kafka")\
.option("kafka.bootstrap.servers", kafka_bootstrap_server).save()

processedDF = dataframe.select(processedSchema())
processedDF.write.format("csv").mode("append").option("sep","^").option("compression","gzip").option("path", cloud_storage_path).save()

[Kaniz_Fatma]

Hi @Rajdeepak, To address data redundancy issues caused by reprocessing during application restarts, consider these strategies: Ensure proper checkpointing by configuring and protecting your checkpoint directory; manage Kafka offsets correctly by setting `enable.auto.commit` to `true` and optionally managing them manually; make write operations idempotent to prevent duplicates by using unique identifiers; handle failures gracefully with robust error handling and retry mechanisms; and optimize checkpointing and state management by tuning intervals and maintaining state stores. Adjust your Kafka read stream and write stream configurations accordingly to ensure correct offset management and checkpointing setup.

For more detailed guidance, you can refer to the following resources:

• Preventing Data Reprocessing in Spark Streaming
• Dynamic ETL with PySpark

Feel free to ask if you need further assistance or have more questions!