How to migrate kafka consumer offset information in spark - Stack Overflow

admin管理员组

文章数量:1025783

I have spark job that reads messages from kafka , using

spark.readStream(.....)

method. This is deployed in AWS as sparkapp and it reads data from MSK kafka. Recently the devOPs decided to migrate to a different MSK cluster for cost reduction.

There were some go Lang services which use their kafka offsets maintained by kafka server itself and the all the consumer offsets were automaticallly migrated to new MSK cluster since it is being managed by kafka itself.

However for spark jobs we have structured streaming which has checkpointing which is being managed by spark job itself. So when this spark job is being made to listen to new kafka URLs , they fail with missing offsets error. To resolve this we need to set failOnDataLoss = false , or delete checkpoint, in either case we have data loss and state gets lost, if the job has some custom state logic .

I have 2 requirements

1. Is there a way that kafka offsets are migrated automatically for spark consumers too whenever there is a migration of MSK, assuming that spark job has structured streaming.

2. Is there a way that we can delete checkpoint, so that we can remove the offsets tracked by checkpoint, in order to reprocess with "earliest" offset, however without losing the Spark state .

I have spark job that reads messages from kafka , using

spark.readStream(.....)

method. This is deployed in AWS as sparkapp and it reads data from MSK kafka. Recently the devOPs decided to migrate to a different MSK cluster for cost reduction.

There were some go Lang services which use their kafka offsets maintained by kafka server itself and the all the consumer offsets were automaticallly migrated to new MSK cluster since it is being managed by kafka itself.

However for spark jobs we have structured streaming which has checkpointing which is being managed by spark job itself. So when this spark job is being made to listen to new kafka URLs , they fail with missing offsets error. To resolve this we need to set failOnDataLoss = false , or delete checkpoint, in either case we have data loss and state gets lost, if the job has some custom state logic .

I have 2 requirements

1. Is there a way that kafka offsets are migrated automatically for spark consumers too whenever there is a migration of MSK, assuming that spark job has structured streaming.

2. Is there a way that we can delete checkpoint, so that we can remove the offsets tracked by checkpoint, in order to reprocess with "earliest" offset, however without losing the Spark state .

本文标签： How to migrate kafka consumer offset information in sparkStack Overflow