spark与kafaka整合workcount示例分析

今天就跟大家聊聊有关spark与kafaka整合workcount示例分析，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

package hgs.spark.streamingimport org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport org.apache.spark.streaming.StreaminGContextimport org.apache.spark.streaming.Secondsimport org.apache.spark.streaming.kafka.KafkaUtilsimport org.apache.spark.storage.StorageLevelimport kafka.serializer.StringDecoderimport org.apache.kafka.common.serialization.StringDeserializerimport kafka.serializer.DefaultDecoderimport org.apache.spark.HashPartitionerobject SparkStreamingKafkaReciverWordCount {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")    val sc = new SparkContext(conf)    val ssc = new StreamingContext(sc,Seconds(4))     ssc.checkpoint("d:\\checkpoint")        val updateFunc=(iter:Iterator[(String,Seq[Int],Option[Int])])=>{    //iter.flatMap(it=>Some(it._2.sum+it._3.getOrElse(0)).map((it._1,_)))//方式一    //iter.flatMap{case(x,y,z)=>{Some(y.sum+z.getOrElse(0)).map((x,_))}}//方式二    iter.flatMap(it=>Some(it._1,(it._2.sum.toInt+it._3.getOrElse(0))))//方式三    }    //注意下面的map一定要加上泛型，否则createStream会报错    //kafaka的一些参数    val props = Map[String,String](             "bootstrap.servers"->"bigdata01:9092,bigdata02:9092,bigdata03:9092",             "group.id"->"group_test",             "enable.auto.commit"->"true",             "auto.commit.intervals.ms"->"2000",             "auto.offset.reset"->"smallest",             "ZooKeeper.connect"->"bigdata01:2181,bigdata02:2181,bigdata03:2181")    //topics    val topics = Map[String,Int]("test"->1)        val rds = KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](ssc, props, topics, StorageLevel.MEMORY_AND_DISK)        val words = rds.flatMap(x=>x._2.split(" "))    val wordscount = words.map((_,1)).updateStateByKey(updateFunc, new HashPartitioner(sc.defaultMinPartitions), true)        wordscount.print()    //启动    ssc.start()    ssc.awaitTermination()      }}

看完上述内容，你们对spark与kafaka整合workcount示例分析有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注编程网精选频道，感谢大家的支持。