Building Data Streaming Applications with Apache Kafka

Scala example for receiver-based integration

Here is an example on Scala:

import org.apache.Spark.SparkConfimport org.apache.Spark.streaming.{Minutes, Seconds, StreamingContext}import org.apache.Spark.streaming.kafka._object KafkaWordCount {  def main(args: Array[String]) {    val zkQuorum:String="localhost:2181"    val group:String="stream"    val numThreads:String="3"    val topics:String="test"    val SparkConf = new SparkConf().setAppName("KafkaWordCount")    val ssc = new StreamingContext(SparkConf, Seconds(2))    ssc.checkpoint("WALCheckpoint")    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap    val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)    val words = lines.flatMap(_.split(" "))    val wordCounts = words.map(x ...