SparkStreaming 整合Kafka

Spark Streaming连接kafka 的两种方式

Receiver based Approah

KafkaUtils.createDstream基于接收器方式，消费Kafka数据，已淘汰Receiver作为Task运行在Executor等待数据，一个Receiver效率低，需要开启多个，再手动合并数据，很麻烦Receiver挂了，可能丢失数据，需要开启WAL(预写日志)保证数据安全，效率低通过Zookeeper来连接kafka，offset存储再zookeeper中spark消费的时候为了保证数据不丢也会保存一份offset，可能出现数据不一致

Direct Approach

KafkaUtils.createDirectStream直连方式，streaming中每个批次的job直接调用Simple Consumer API获取对应Topic数据Direct方式直接连接kafka分区获取数据，提高了并行能力Direct方式调用kafka低阶API，offset自己存储和维护，默认由spark维护在checkpoint中offset也可以自己手动维护，保存在mysql/redis中

// 从kafka加载数据val kafkaParams = Map[String, Object](  "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址  "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则  "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则  "group.id" -> "sparkdemo",//消费者组名称  //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费  //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费  //none:表示如果有offset记录从offset记录开始消费,如果没有就报错  "auto.offset.reset" -> "latest",  "auto.commit.interval.ms"->"1000",//自动提交的时间间隔  "enable.auto.commit" -> (true: java.lang.Boolean)//是否自动提交)val topics = Array("spark_kafka")//要订阅的主题//使用工具类从Kafka中消费消息val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](  ssc,  LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的)

代码展示

自动提交偏移量

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加载数据    val kafkaParams = Map[String, Object](      "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址      "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则      "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则      "group.id" -> "sparkdemo",//消费者组名称      //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费      //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费      //none:表示如果有offset记录从offset记录开始消费,如果没有就报错      "auto.offset.reset" -> "latest",      "auto.commit.interval.ms"->"1000",//自动提交的时间间隔      "enable.auto.commit" -> (true: java.lang.Boolean)//是否自动提交    )    val topics = Array("spark_kafka")//要订阅的主题    //使用工具类从Kafka中消费消息    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](      ssc,      LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的    )    // 处理消息    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)  }}

手动提交

提交代码

// 处理消息//注意提交的时机:应该是消费完一小批就该提交一次offset,而在DStream一小批的体现是RDDkafkaDS.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("消费" + info)  })  //获取rdd中offset相关的信息:offsetRanges里面就包含了该批次各个分区的offset信息  val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  //提交  kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)  println("当前批次的数据已消费并手动提交")})

完整代码

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加载数据    val kafkaParams = Map[String, Object](      "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址      "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则      "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则      "group.id" -> "sparkdemo",//消费者组名称      //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费      //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费      //none:表示如果有offset记录从offset记录开始消费,如果没有就报错      "auto.offset.reset" -> "latest",//      "auto.commit.interval.ms"->"1000",//自动提交的时间间隔      "enable.auto.commit" -> (false: java.lang.Boolean)//是否自动提交    )    val topics = Array("spark_kafka")//要订阅的主题    //使用工具类从Kafka中消费消息    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](      ssc,      LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的    )    // 处理消息    //注意提交的时机:应该是消费完一小批就该提交一次offset,而在DStream一小批的体现是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("消费" + info)      })      //获取rdd中offset相关的信息:offsetRanges里面就包含了该批次各个分区的offset信息      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges      //提交      kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)      println("当前批次的数据已消费并手动提交")    })    // 输出    kafkaDS.print()    ssc.start()    ssc.awaitTermination()    ssc.stop(true, true)  }}