Spark Streaming указывает начальные и конечные смещения

Question

У меня есть сценарий, в котором я хочу повторно обработать определенный пакет данных, поступающих из Kafka, используя Spark DStreams.

допустим, я хочу повторно обработать следующие пакеты данных.

Topic-Partition1- {1000,2000} Topic-Partition2- {500-600}

Ниже приведен фрагмент кода, в котором я могу указать начальные смещения.

val inputDStream = KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Assign[String, String](
      topic-partition-list,
      kafkaProps,
      starting-offset-ranges))

Но, я хочу знать, в любом случае, я могу также указать конечные смещения, как в случае структурированной потоковой передачипакетный режим.

По сути, он должен обработать этот небольшой пакет и остановить рабочий процесс.

Примечание: я не хочу использовать структурированную потоковую передачу для этого варианта использования.Хотите использовать только DStreams.

Answer 1

Нашли способ сделать это.

val offsetRanges = Array(
  // topic, partition, inclusive starting offset, exclusive ending offset
  OffsetRange("test", 0, 0, 100),
  OffsetRange("test", 1, 0, 100)
)

val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)