Receiver-based Streaming

/**
 * Create an input stream that pulls messages from Kafka Brokers using receivers.
 *
 * @param ssc StreamingContext object
 * @param zkQuorum Zookeeper quorum (hostname:port,hostname:port,..)
 * @param groupId The group id for this consumer
 * @param topics Map of (topic_name -> numPartitions) to consume. Each partition is consumed
 *               in its own thread
 * @param storageLevel Storage level to use for storing the received objects
 *                     (default: StorageLevel.MEMORY_AND_DISK_SER_2)
 * @return DStream of (Kafka message key, Kafka message value)
 */
def createStream(
  ssc: StreamingContext,
  zkQuorum: String,
  groupId: String,
  topics: Map[String, Int],
  storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[(String, String)]

import org.apache.spark.streaming.kafka._
import org.apache.spark.storage.StorageLevel

val zkQuorum = "localhost:2181"
val groupId = "my-consumer-group"
val topics = Map("user-events" -> 1, "purchase-events" -> 2)

val stream = KafkaUtils.createStream(
  streamingContext, zkQuorum, groupId, topics, StorageLevel.MEMORY_AND_DISK_SER_2
)

stream.foreachRDD { rdd =>
  rdd.foreach { case (key, value) =>
    println(s"Received: $key -> $value")
  }
}

/**
 * Create an input stream that pulls messages from Kafka Brokers using receivers
 * with custom key/value types.
 *
 * @param ssc StreamingContext object
 * @param kafkaParams Map of kafka configuration parameters
 * @param topics Map of (topic_name -> numPartitions) to consume
 * @param storageLevel Storage level to use for storing the received objects
 * @tparam K type of Kafka message key
 * @tparam V type of Kafka message value
 * @tparam U type of Kafka message key decoder
 * @tparam T type of Kafka message value decoder
 * @return DStream of (Kafka message key, Kafka message value)
 */
def createStream[K: ClassTag, V: ClassTag, U <: Decoder[K]: ClassTag, T <: Decoder[V]: ClassTag](
  ssc: StreamingContext,
  kafkaParams: Map[String, String],
  topics: Map[String, Int],
  storageLevel: StorageLevel
): ReceiverInputDStream[(K, V)]

import kafka.serializer.{StringDecoder, DefaultDecoder}

val kafkaParams = Map[String, String](
  "zookeeper.connect" -> "localhost:2181",
  "group.id" -> "my-consumer-group",
  "zookeeper.connection.timeout.ms" -> "10000"
)

val topics = Map("binary-data" -> 1)

val binaryStream = KafkaUtils.createStream[Array[Byte], Array[Byte], DefaultDecoder, DefaultDecoder](
  streamingContext, kafkaParams, topics, StorageLevel.MEMORY_AND_DISK_SER
)

binaryStream.foreachRDD { rdd =>
  rdd.foreach { case (keyBytes, valueBytes) =>
    println(s"Key length: ${keyBytes.length}, Value length: ${valueBytes.length}")
  }
}

/**
 * Create an input stream that pulls messages from Kafka Brokers using receivers (Java API).
 *
 * @param jssc JavaStreamingContext object
 * @param zkQuorum Zookeeper quorum (hostname:port,hostname:port,..)
 * @param groupId The group id for this consumer
 * @param topics Map of (topic_name -> numPartitions) to consume
 * @return DStream of (Kafka message key, Kafka message value)
 */
public static JavaPairReceiverInputDStream<String, String> createStream(
  JavaStreamingContext jssc,
  String zkQuorum,
  String groupId,
  Map<String, Integer> topics
)

/**
 * Create an input stream with custom storage level (Java API).
 */
public static JavaPairReceiverInputDStream<String, String> createStream(
  JavaStreamingContext jssc,
  String zkQuorum,
  String groupId,
  Map<String, Integer> topics,
  StorageLevel storageLevel
)

import org.apache.spark.streaming.kafka.KafkaUtils;
import org.apache.spark.storage.StorageLevel;

Map<String, Integer> topics = new HashMap<>();
topics.put("my-topic", 1);

JavaPairReceiverInputDStream<String, String> stream = KafkaUtils.createStream(
    jssc,
    "localhost:2181",
    "my-consumer-group", 
    topics,
    StorageLevel.MEMORY_AND_DISK_SER_2()
);

stream.foreachRDD(rdd -> {
    rdd.foreach(record -> {
        System.out.println("Key: " + record._1 + ", Value: " + record._2);
    });
});

/**
 * Create an input stream with custom types and decoders (Java API).
 *
 * @param jssc JavaStreamingContext object
 * @param keyTypeClass Key type of DStream
 * @param valueTypeClass Value type of DStream
 * @param keyDecoderClass Type of kafka key decoder
 * @param valueDecoderClass Type of kafka value decoder
 * @param kafkaParams Map of kafka configuration parameters
 * @param topics Map of (topic_name -> numPartitions) to consume
 * @param storageLevel RDD storage level
 * @tparam K type of Kafka message key
 * @tparam V type of Kafka message value
 * @tparam U type of Kafka message key decoder
 * @tparam T type of Kafka message value decoder
 * @return DStream of (Kafka message key, Kafka message value)
 */
public static <K, V, U extends Decoder<K>, T extends Decoder<V>> 
JavaPairReceiverInputDStream<K, V> createStream(
  JavaStreamingContext jssc,
  Class<K> keyTypeClass,
  Class<V> valueTypeClass,
  Class<U> keyDecoderClass,
  Class<T> valueDecoderClass,
  Map<String, String> kafkaParams,
  Map<String, Integer> topics,
  StorageLevel storageLevel
)

StorageLevel.MEMORY_AND_DISK_SER_2    // Default: Memory + disk, serialized, replicated
StorageLevel.MEMORY_AND_DISK_SER      // Memory + disk, serialized, not replicated  
StorageLevel.MEMORY_ONLY_SER_2        // Memory only, serialized, replicated
StorageLevel.DISK_ONLY_2              // Disk only, replicated

Map(
  "zookeeper.connect" -> "localhost:2181",
  "group.id" -> "my-consumer-group",
  "zookeeper.connection.timeout.ms" -> "10000",
  "zookeeper.session.timeout.ms" -> "10000",
  "zookeeper.sync.time.ms" -> "2000",
  "auto.commit.interval.ms" -> "1000"
)

// Map of topic name to number of threads/partitions to consume
Map(
  "topic1" -> 1,      // 1 thread for topic1
  "topic2" -> 2,      // 2 threads for topic2  
  "topic3" -> 4       // 4 threads for topic3
)

// Old receiver-based approach
val receiverStream = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)

// New direct streaming approach  
val directStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topics.keySet
)