快捷導(dǎo)航

SparkStreaming整合Kafka過程詳解

更新時間：2023年01月27日 10:48:24 作者：健鑫.

這篇文章主要介紹了SparkStreaming整合Kafka過程，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧

Spark Streaming連接kafka 的兩種方式

Receiver based Approah

KafkaUtils.createDstream基于接收器方式，消費(fèi)Kafka數(shù)據(jù)，已淘汰
Receiver作為Task運(yùn)行在Executor等待數(shù)據(jù)，一個Receiver效率低，需要開啟多個，再手動合并數(shù)據(jù)，很麻煩
Receiver掛了，可能丟失數(shù)據(jù)，需要開啟WAL(預(yù)寫日志)保證數(shù)據(jù)安全，效率低
通過Zookeeper來連接kafka，offset存儲再zookeeper中
spark消費(fèi)的時候?yàn)榱吮ＷC數(shù)據(jù)不丟也會保存一份offset，可能出現(xiàn)數(shù)據(jù)不一致

Direct Approach

KafkaUtils.createDirectStream直連方式，streaming中每個批次的job直接調(diào)用Simple Consumer API獲取對應(yīng)Topic數(shù)據(jù)
Direct方式直接連接kafka分區(qū)獲取數(shù)據(jù)，提高了并行能力
Direct方式調(diào)用kafka低階API，offset自己存儲和維護(hù)，默認(rèn)由spark維護(hù)在checkpoint中
offset也可以自己手動維護(hù)，保存在mysql/redis中

// 從kafka加載數(shù)據(jù)
val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址
  "key.deserializer" -> classOf[StringDeserializer],//key的反序列化規(guī)則
  "value.deserializer" -> classOf[StringDeserializer],//value的反序列化規(guī)則
  "group.id" -> "sparkdemo",//消費(fèi)者組名稱
  //earliest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最早的消息開始消費(fèi)
  //latest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最后/最新的消息開始消費(fèi)
  //none:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有就報(bào)錯
  "auto.offset.reset" -> "latest",
  "auto.commit.interval.ms"->"1000",//自動提交的時間間隔
  "enable.auto.commit" -> (true: java.lang.Boolean)//是否自動提交
)
val topics = Array("spark_kafka")//要訂閱的主題
//使用工具類從Kafka中消費(fèi)消息
val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent, //位置策略,使用源碼中推薦的
  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消費(fèi)策略,使用源碼中推薦的
)

代碼展示

自動提交偏移量

object kafka_Demo01 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("kafka_Demo01")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc, Seconds(5))
    ssc.checkpoint("data/ckp")
    // 從kafka加載數(shù)據(jù)
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址
      "key.deserializer" -> classOf[StringDeserializer],//key的反序列化規(guī)則
      "value.deserializer" -> classOf[StringDeserializer],//value的反序列化規(guī)則
      "group.id" -> "sparkdemo",//消費(fèi)者組名稱
      //earliest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最早的消息開始消費(fèi)
      //latest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最后/最新的消息開始消費(fèi)
      //none:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有就報(bào)錯
      "auto.offset.reset" -> "latest",
      "auto.commit.interval.ms"->"1000",//自動提交的時間間隔
      "enable.auto.commit" -> (true: java.lang.Boolean)//是否自動提交
    )
    val topics = Array("spark_kafka")//要訂閱的主題
    //使用工具類從Kafka中消費(fèi)消息
    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //位置策略,使用源碼中推薦的
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消費(fèi)策略,使用源碼中推薦的
    )
    // 處理消息
    val infoDS = kafkaDS.map(record => {
      val topic = record.topic()
      val partition = record.partition()
      val offset = record.offset()
      val key = record.key()
      val value = record.value()
      val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}"""
      info
    })
    // 輸出
    infoDS.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(true, true)
  }
}

手動提交

提交代碼

// 處理消息
//注意提交的時機(jī):應(yīng)該是消費(fèi)完一小批就該提交一次offset,而在DStream一小批的體現(xiàn)是RDD
kafkaDS.foreachRDD(rdd => {
  rdd.foreach(record => {
    val topic = record.topic()
    val partition = record.partition()
    val offset = record.offset()
    val key = record.key()
    val value = record.value()
    val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}"""
    info
    println("消費(fèi)" + info)
  })
  //獲取rdd中offset相關(guān)的信息:offsetRanges里面就包含了該批次各個分區(qū)的offset信息
  val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  //提交
  kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  println("當(dāng)前批次的數(shù)據(jù)已消費(fèi)并手動提交")
})

完整代碼

object kafka_Demo02 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("kafka_Demo01")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc, Seconds(5))
    ssc.checkpoint("data/ckp")
    // 從kafka加載數(shù)據(jù)
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址
      "key.deserializer" -> classOf[StringDeserializer],//key的反序列化規(guī)則
      "value.deserializer" -> classOf[StringDeserializer],//value的反序列化規(guī)則
      "group.id" -> "sparkdemo",//消費(fèi)者組名稱
      //earliest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最早的消息開始消費(fèi)
      //latest:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有從最后/最新的消息開始消費(fèi)
      //none:表示如果有offset記錄從offset記錄開始消費(fèi),如果沒有就報(bào)錯
      "auto.offset.reset" -> "latest",
//      "auto.commit.interval.ms"->"1000",//自動提交的時間間隔
      "enable.auto.commit" -> (false: java.lang.Boolean)//是否自動提交
    )
    val topics = Array("spark_kafka")//要訂閱的主題
    //使用工具類從Kafka中消費(fèi)消息
    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //位置策略,使用源碼中推薦的
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消費(fèi)策略,使用源碼中推薦的
    )
    // 處理消息
    //注意提交的時機(jī):應(yīng)該是消費(fèi)完一小批就該提交一次offset,而在DStream一小批的體現(xiàn)是RDD
    kafkaDS.foreachRDD(rdd => {
      rdd.foreach(record => {
        val topic = record.topic()
        val partition = record.partition()
        val offset = record.offset()
        val key = record.key()
        val value = record.value()
        val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}"""
        info
        println("消費(fèi)" + info)
      })
      //獲取rdd中offset相關(guān)的信息:offsetRanges里面就包含了該批次各個分區(qū)的offset信息
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      //提交
      kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      println("當(dāng)前批次的數(shù)據(jù)已消費(fèi)并手動提交")
    })
    // 輸出
    kafkaDS.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(true, true)
  }
}

到此這篇關(guān)于SparkStreaming整合Kafka過程詳解的文章就介紹到這了,更多相關(guān)SparkStreaming整合Kafka內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: