Мне нужно хранить и извлекать смещения Кафки из задания Spark Dataframe в пакетном режиме , используя опцию startingOffsets.
По сути, я хочу, чтобы каждый раз, когда задание выполнялось, предоставлял startOffsets json.начинает использовать традиционный синтаксис:
val json = """{"test":{"0":2,"1":3,"2":1,"3":0}}"""
spark.read
.format("kafka")
.option("kafka.bootstrap.servers", bootstrapServers)
.option("subscribe", topic)
.option("startingOffsets", json)
.load()
Предположим, что тема теста имеет 3 раздела (0,1,2) .
Я хочудля этого (скажите, пожалуйста, есть ли лучший способ):
- Каждый раз, когда начинается работа, читается в файле initialOffsets , сохраненном в HDFS
- В конце каждого прогона я хочу сохранить новый файл startOffsets с обновленными смещениями?Как я могу это сделать?Я увидел, что схема довольно странная, и я думаю, что мне нужно построить ее программно , потому что она имеет тип структуры внутри с произвольным числом полей, в зависимости от количества разделов, например:
Проверка темы с 2 и 3 разделами:
StructType(StructField("test",StructType(StructField(0,LongType,true), StructField(1,LongType,true)),true))
StructType(StructField("test",StructType(StructField(0,LongType,true), StructField(1,LongType,true),StructField(2,LongType,true)),true))
- Что если после двух или трех запусков тема разделена, например, на 4 раздела?В этом случае Spark жалуется мне:
Если начальные смещения содержат определенные смещения, вы должны указать все разделы темы.
В этом случае я думал, что читать с помощью KafkaProducer API количество разделов и проверьте, равно ли количество предыдущих разделов.Если они отличаются от меня, нужно манипулировать структурой схемы .Кажется сложным.Есть еще один способ наверняка.
- Есть ли какой-нибудь простой и приятный способ произвести запуск startOffsets json для сохранения с помощью схемы выше?