SaprkStreaming整合kafka（获取偏移量和异步提交偏移量）

131 阅读 0 评论 87 点赞

我是靠谱客的博主任性芒果，最近开发中收集的这篇文章主要介绍SaprkStreaming整合kafka（获取偏移量和异步提交偏移量），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

package cn._51doit.spark.day13

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object KafkaWordCount {

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

    //实时计算创建StreamingContext，（StreamingContext是对SparkContext的增强包装，里面持有者SparkContext的引用）
    val conf = new SparkConf().setAppName("StreamingWordCount").setMaster("local[*]")

    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")

    //Kafka相关的参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "node-1.51doit.com:9092,node-2.51doit.com:9092,node-3.51doit.com:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "g02",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean) //让消费者不用自动提交偏移量
    )

    val topics = Array("wordcount")

    //sparkStreaming跟Kafka整合，使用的是官方推荐的直连方式，使用Kafka底层的消费API，效率更高
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc, //传入StreamingContext
      LocationStrategies.PreferConsistent, //位置策略
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略
    )

    //foreachRDD既不是Transformation也不是Action
    //（foreachRDD中的函数也会定期的调用，传入到foreachRDD中的函数，是在Driver调用的）
    kafkaDStream.foreachRDD(rdd => {
      if(!rdd.isEmpty()) {

        //获取偏移量,第一手的RDD是特殊的RDD（KafkaRDD），只有KafkaRDD有偏移量(Driver端获取的)
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.map(_.value())
          .flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKey(_ + _)
          .foreach(println)

        //（在Driver端异步的更新偏移量,将偏移量写入到Kafka特殊的Topic中，__consumer_offset）
        kafkaDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) //在Driver提交的

      }
    })


    //开启
    ssc.start()
    //让程序一直运行，将Driver挂起
    ssc.awaitTermination()

  }
}

kafka参数中设置了"enable.auto.commit" -> (false: java.lang.Boolean) ，让消费者不用自动提交偏移量，所以我们需要自己获取偏移量并提交（不设置这行代码或者false改为true，就是kafka自动提交偏移量，也是默认的方式）

偏移量信息只存在于第一手的DStream中，就是上面代码中的kafkaDstream，类型是InputStream。
要获取偏移量，需要调用foreachRDD（foreachRDD不是action，也不是tranformation，是在diver端执行的）foreachRDD的作用就是把kafkaDStream中的每个批次的RDD周期性的取出来，简单说就是一个遍历的作用。