Batch Processing

def createRDD[K, V](
  sc: SparkContext,
  kafkaParams: java.util.Map[String, Object],
  offsetRanges: Array[OffsetRange],
  locationStrategy: LocationStrategy
): RDD[ConsumerRecord[K, V]]

public static <K, V> JavaRDD<ConsumerRecord<K, V>> createRDD(
  JavaSparkContext jsc,
  java.util.Map<String, Object> kafkaParams,
  OffsetRange[] offsetRanges,
  LocationStrategy locationStrategy
)

import org.apache.spark.streaming.kafka010._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.TopicPartition

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "batch-processing-group"
)

// Define specific offset ranges to process
val offsetRanges = Array(
  OffsetRange("orders", 0, 100, 200),      // Process messages 100-199 from orders partition 0
  OffsetRange("orders", 1, 50, 150),       // Process messages 50-149 from orders partition 1
  OffsetRange("payments", 0, 0, 100)       // Process messages 0-99 from payments partition 0
)

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

// Process the RDD
val processedData = rdd.map { record =>
  (record.topic, record.partition, record.offset, record.key, record.value)
}.collect()

processedData.foreach { case (topic, partition, offset, key, value) =>
  println(s"Topic: $topic, Partition: $partition, Offset: $offset, Key: $key, Value: $value")
}

import org.apache.spark.streaming.kafka010._

// Get offset ranges from a previous streaming batch
val previousOffsetRanges: Array[OffsetRange] = // ... obtained from HasOffsetRanges

// Create RDD to reprocess the same data
val reprocessRDD = KafkaUtils.createRDD[String, String](
  sparkContext,
  kafkaParams,
  previousOffsetRanges,
  LocationStrategies.PreferConsistent
)

// Apply different processing logic
val reprocessedResults = reprocessRDD
  .filter(record => record.value.contains("error"))
  .map(record => s"Reprocessed: ${record.value}")
  .collect()

import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "localhost:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "batch-processing-group");

OffsetRange[] offsetRanges = {
  OffsetRange.create("topic1", 0, 100L, 200L),
  OffsetRange.create("topic1", 1, 150L, 250L)
};

JavaRDD<ConsumerRecord<String, String>> rdd = KafkaUtils.createRDD(
  javaSparkContext,
  kafkaParams,
  offsetRanges,
  LocationStrategies.PreferConsistent()
);

// Process the RDD
rdd.foreach(record -> {
  System.out.println("Topic: " + record.topic() + 
                    ", Partition: " + record.partition() + 
                    ", Offset: " + record.offset() + 
                    ", Value: " + record.value());
});

import org.apache.spark.streaming.kafka010._

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

// RDD implements HasOffsetRanges, so you can get offset information
val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

ranges.foreach { range =>
  println(s"Topic: ${range.topic}, Partition: ${range.partition}, " +
          s"From: ${range.fromOffset}, Until: ${range.untilOffset}, Count: ${range.count}")
}

// Process data and track progress
val processedCount = rdd.count()
val totalMessages = ranges.map(_.count).sum
println(s"Processed $processedCount messages out of $totalMessages total")