object KafkaWordCount{

    val updateFunc = (iter:Iterator[(String,Seq[Int],Option[Int])])=>{

       iter.flatMap{case(x,y,z) => Some(y.sum+z.getOrElse(0)).map(i => (x,i))}

    }

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

       val Array(zkQuorum, group, topics, numThreads) = args

       val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")

       val ssc = new StreamingContext(sparkConf, Seconds(5))

       ssc.checkpoint("c://ck2")

       val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

       val data = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER)

       val words = data.map(_._2).flatMap(_.split(" "))

       val wordCounts = words.map((_, 1)).updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)

       println(wordCounts)

       ssc.start()

       ssc.awaitTermination()

    }

 }

 17/07/28 17:11:59 ERROR StreamingContext: Error starting the context, marking it as stopped

 java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute

     at scala.Predef$.require(Predef.scala:233)

     at org.apache.spark.streaming.DStreamGraph.validate(DStreamGraph.scala:161)

     at org.apache.spark.streaming.StreamingContext.validate(StreamingContext.scala:542)

     at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:601)

     at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:600)

     at org.bianqi.spark.day5.KafkaWordCount$.main(KafkaWordCount.scala:24)

     at org.bianqi.spark.day5.KafkaWordCount.main(KafkaWordCount.scala)

 Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute

     at scala.Predef$.require(Predef.scala:233)

 object KafkaWordCount{

    val updateFunc = (iter:Iterator[(String,Seq[Int],Option[Int])])=>{

       iter.flatMap{case(x,y,z) => Some(y.sum+z.getOrElse(0)).map(i => (x,i))}

    }

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

       val Array(zkQuorum, group, topics, numThreads) = args

       val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")

       val ssc = new StreamingContext(sparkConf, Seconds(5))

       ssc.checkpoint("c://ck2")

       val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

       val data = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER)

       val words = data.map(_._2).flatMap(_.split(" "))

       val wordCounts = words.map((_, 1)).updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)

       wordCounts.print()

       ssc.start()

       ssc.awaitTermination()

    }

 }