Как включить и «последний», и «JSON с определенным смещением» в «startOffsets» при импорте данных из Kafka в Spark Structured Streaming

Question

У меня есть потоковый запрос, сохраняющий данные в fileink. Я использую .option («стартовый набор», «последний») и местоположение контрольной точки. Если в Spark есть какое-то время простоя, и когда потоковый запрос начинается снова, я не хочу начинать обработку с того места, где запрос был остановлен, когда он завершился, а не в этом сценарии, я также хотел бы добавить ("initialOffsets", "" "{ "topicA": {"0": 23, "1": - 1}, "topicB": {"0": - 2}} "" "), указав смещение, определяемое пользователем, которое необходимо обработать.

Я пытался сделать это с разными программами, но мне нужно добиться этого в одной программе

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger

object OSB_offset_kafkaToSpark {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.
      builder().
      appName("OSB_kafkaToSpark").
      config("spark.mongodb.output.uri", "spark.mongodb.output.uri=mongodb://somemongodb.com:27018").
      getOrCreate()

    println("SparkSession -> "+spark)

    import spark.implicits._

    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "somekafkabroker:9092, somekafkabroker:9092")
      .option("subscribe", "someTopic")
      .option("startingOffsets", "latest")
      .option("startingOffsets",""" {"someTopic":{"0":438521}}, "someTopic":{"1":438705}}, "someTopic":{"2":254180}}""")
      .option("endingOffsets",""" {"someTopic":{"0":-1}}, "someTopic":{"1":-1}}, "someTopic":{"2":-1}} """)
      .option("failOnDataLoss", "false")
      .load()
    val dfs = df.selectExpr("CAST(value AS STRING)")

    val data = dfs.withColumn("splitted", split($"value", "/"))
      .select($"splitted".getItem(4).alias("region"), $"splitted".getItem(5).alias("service"), col("value"))
      .withColumn("service_type", regexp_extract($"service", """.*(Inbound|Outbound|Outound).*""", 1))
      .withColumn("region_type", concat(
        when(col("region").isNotNull, col("region")).otherwise(lit("null")), lit(" "),
        when(col("service").isNotNull, col("service_type")).otherwise(lit("null"))))
      .withColumn("datetime", regexp_extract($"value", """\d{4}-[01]\d-[0-3]\d [0-2]\d:[0-5]\d:[0-5]\d""", 0))

    val extractedDF = data.filter(
      col("region").isNotNull &&
        col("service").isNotNull &&
        col("value").isNotNull &&
        col("service_type").isNotNull &&
        col("region_type").isNotNull &&
        col("datetime").isNotNull)
      .filter("region != ''")
      .filter("service != ''")
      .filter("value != ''")
      .filter("service_type != ''")
      .filter("region_type != ''")
      .filter("datetime != ''")

    val pathstring = "/user/spark_streaming".concat(args(0))


    val query = extractedDF.writeStream
      .format("json")
      .option("path", pathstring)
      .option("checkpointLocation", "/user/some_checkpoint")
      .outputMode("append")
      .trigger(Trigger.ProcessingTime("5 seconds"))
      .start()

    query.awaitTermination()

  }
}

Мне нужно запустить одну программу с .option ("initialOffsets", "latest") и .option ("initialOffsets", "" "{" someTopic ": {" 0 ": 438521}}," someTopic " : {"1": 438705}}, "someTopic": {"2": 254180}} "" ").

Я не уверен, достижимо ли это