spark-streaming-kafka怎樣通過KafkaUtils.createDirectStream的方式處理數(shù)據(jù)，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

10年積累的成都做網(wǎng)站、成都網(wǎng)站建設經(jīng)驗，可以快速應對客戶對網(wǎng)站的新想法和需求。提供各種問題對應的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡服務。我雖然不認識你，你也不認識我。但先網(wǎng)站制作后付款的網(wǎng)站建設流程，更有松桃免費網(wǎng)站建設讓你可以放心的選擇與我們合作。

package hgs.spark.streaming
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.kafka.KafkaCluster
import scala.collection.immutable.Map
import java.util.NoSuchElementException
import org.apache.spark.SparkException
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import org.codehaus.jackson.map.deser.std.PrimitiveArrayDeserializers.StringDeser
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.DirectKafkaInputDStream
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.kafka.HasOffsetRanges
import org.apache.spark.HashPartitioner
object SparkStreamingKafkaDirectWordCount {
  def main(args: Array[String]): Unit = {
     val conf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[5]")
     conf.set("spark.streaming.kafka.maxRatePerPartition", "1")
     val sc = new SparkContext(conf)
     val ssc = new StreamingContext(sc,Seconds(1)) 
     ssc.checkpoint("d:\\checkpoint")
     val kafkaParams = Map[String,String](
         "metadata.broker.list"->"bigdata01:9092,bigdata02:9092,bigdata03:9092",
         "group.id"->"group_hgs",
         "zookeeper.connect"->"bigdata01:2181,bigdata02:2181,bigdata03:2181")
     val kc = new KafkaCluster(kafkaParams)
     val topics = Set[String]("test")
     //每個rdd返回的數(shù)據(jù)是(K,V)類型的，該函數(shù)規(guī)定了函數(shù)返回數(shù)據(jù)的類型
     val mmdFunct = (mmd: MessageAndMetadata[String, String])=>(mmd.topic+" "+mmd.partition,mmd.message())
     
     val rds = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder,(String,String)](ssc, kafkaParams, getOffsets(topics,kc,kafkaParams),mmdFunct)
     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))))//方式三
      }
     val words = rds.flatMap(x=>x._2.split(" ")).map((_,1))
     //val wordscount = words.map((_,1)).updateStateByKey(updateFunc, new HashPartitioner(sc.defaultMinPartitions), true)
     //println(getOffsets(topics,kc,kafkaParams))
     rds.foreachRDD(rdd=>{
       if(!rdd.isEmpty()){
         //對每個dataStreamoffset進行更新
         upateOffsets(topics,kc,rdd,kafkaParams)
       }
     }   
    )
     
     words.print()
     ssc.start()
     ssc.awaitTermination()
     
    
  }
  
  
  
  def getOffsets(topics : Set[String],kc:KafkaCluster,kafkaParams:Map[String,String]):Map[TopicAndPartition, Long]={
   
    val topicAndPartitionsOrNull =  kc.getPartitions(topics)
    if(topicAndPartitionsOrNull.isLeft){
      throw new SparkException(s"$topics in the set may not found")
    }
    else{
      val topicAndPartitions = topicAndPartitionsOrNull.right.get
      val groups = kafkaParams.get("group.id").get
      val offsetOrNull = kc.getConsumerOffsets(groups, topicAndPartitions)
      if(offsetOrNull.isLeft){
        println(s"$groups you assignment may not exists!now redirect to zero!")
        //如果沒有消費過，則從最開始的位置消費
        val erliestOffset = kc.getEarliestLeaderOffsets(topicAndPartitions)
        if(erliestOffset.isLeft)
          throw new SparkException(s"Topics and Partions not definded not found!")
        else
          erliestOffset.right.get.map(x=>(x._1,x._2.offset))
      }
      else{
        //如果消費組已經(jīng)存在則從記錄的地方開始消費
        offsetOrNull.right.get
      }
    }
    
  }
  
  //每次拉取數(shù)據(jù)后存儲offset到ZK
  def upateOffsets(topics : Set[String],kc:KafkaCluster,directRDD:RDD[(String,String)],kafkaParams:Map[String,String]){
    val offsetRanges =  directRDD.asInstanceOf[HasOffsetRanges].offsetRanges
    for(offr <-offsetRanges){
      val topicAndPartitions = TopicAndPartition(offr.topic,offr.partition)
      val yesOrNo = kc.setConsumerOffsets(kafkaParams.get("group.id").get, Map(topicAndPartitions->offr.untilOffset))
      if(yesOrNo.isLeft){
        println(s"Error when update offset of $topicAndPartitions")
      }
    }
  }
  
  
 
}
/* val conf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
     val sc = new SparkContext(conf)
     val ssc = new StreamingContext(sc,Seconds(4)) 
    val kafkaParams = Map[String,String](
         "metadata.broker.list"->"bigdata01:9092,bigdata02:9092,bigdata03:9092")
     val kc = new KafkaCluster(kafkaParams)
    
     //獲取topic與paritions的信息
     //val tmp = kc.getPartitions(Set[String]("test7"))
     //結(jié)果：topicAndPartitons=Set([test7,0], [test7,1], [test7,2])
     //val topicAndPartitons = tmp.right.get
     //println(topicAndPartitons)
    
    //每個分區(qū)對應的leader信息
    //val tmp = kc.getPartitions(Set[String]("test7"))
    //val topicAndPartitons = tmp.right.get
    //結(jié)果：leadersPerPartitions= Right(Map([test7,0] -> (bigdata03,9092), [test7,1] -> (bigdata01,9092), [test7,2] -> (bigdata02,9092)))
    //val leadersPerPartitions = kc.findLeaders(topicAndPartitons)
    //println(leadersPerPartitions)
    
    //每增加一條消息，對應的partition的offset都會加1,即LeaderOffset(bigdata02,9092,23576)第三個參數(shù)會加一
    //val tmp = kc.getPartitions(Set[String]("test"))
    //val topicAndPartitons = tmp.right.get
    //結(jié)果t=	Right(Map([test7,0] -> LeaderOffset(bigdata03,9092,23568), [test7,2] -> LeaderOffset(bigdata02,9092,23576), [test7,1] -> LeaderOffset(bigdata01,9092,23571)))
    //val  t = kc.getLatestLeaderOffsets(topicAndPartitons)
    // println(t)
    
    
    //findLeader需要兩個參數(shù) topic 分區(qū)編號
    //val tmp = kc.findLeader("test7",0)
    //結(jié)果leader=RightProjection(Right((bigdata03,9092)))
    //val leader = tmp.right
    //val tp = leader.flatMap(x=>{Either.cond(false, None,(x._1,x._2))})  
    
    val tmp = kc.getPartitions(Set[String]("test"))
    val ttp = tmp.right.get
    
    
    while(true){
      try{
      val tp = kc.getConsumerOffsets("group_test1", ttp)
      val maps = tp.right.get
      println(maps)
      Thread.sleep(2000)
      }
      catch{
        case ex:NoSuchElementException=>{println("test")}
      }
        
    }*/

看完上述內(nèi)容，你們掌握spark-streaming-kafka怎樣通過KafkaUtils.createDirectStream的方式處理數(shù)據(jù)的方法了嗎？如果還想學到更多技能或想了解更多相關內(nèi)容，歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝各位的閱讀！

分享標題：spark-streaming-kafka怎樣通過KafkaUtils.createDirectStream的方式處理數(shù)據(jù)
文章URL：http://www.rwnh.cn/article22/jihocc.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供ChatGPT、建站公司、品牌網(wǎng)站建設、服務器托管、定制網(wǎng)站、標簽優(yōu)化

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)