SparkStreaming整合Kafka过程详解

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

这篇文章主要介绍了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一小批的体现是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("当前批次的数据已消费并手动提交")
})

完整代码

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)
  }
}

到此这篇关于SparkStreaming整合Kafka过程详解的文章就介绍到这了,更多相关SparkStreaming整合Kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java的Lombok之@Builder使用总结
这篇文章主要介绍了Java的Lombok之@Builder使用总结,当不使用@Builder注解到类上,创建T1的有参构造函数,入参不仅包括T1中所有的参数,还包括T中所有的参数,T2的属性由T1在有参构造函数中通过调用父类构造器的方式赋初值,需要的朋友可以参考下
2023-12-12
Java实战之在线租房系统的实现
这篇文章主要介绍了利用Java实现的在线租房系统，文中用到了SpringBoot、Redis、MySQL、Vue等技术，文中示例代码讲解详细，需要的可以参考一下
2022-02-02
Java中Controller、Service、Dao/Mapper层的区别与用法
在Java开发中，通常会采用三层架构（或称MVC架构）来划分程序的职责和功能，分别是Controller层、Service层、Dao/Mapper层，本文将详细给大家介绍了三层的区别和用法,需要的朋友可以参考下
2023-05-05
Spring详细讲解FactoryBean接口的使用
这篇文章主要为大家介绍了Spring容器FactoryBean工厂实例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-06-06
Java实现QQ第三方登录的示例代码
这篇文章主要介绍了Java实现QQ第三方登录的示例代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-11-11
mybatis中使用大于小于等于的正确方法
在mybatis中sql是写在xml映射文件中的，如果sql中有一些特殊字符的话，在解析xml文件的时候就会被转义，下面我们就一起来看一下大于小于等于是怎么转义的
2021-04-04
Java生成exe可执行文件
本文主要介绍了Java如何生成exe可执行文件，想了解更多的小伙伴可以借鉴阅读这篇文章
2023-03-03
Spring AI集成DeepSeek实现流式输出的操作方法
本文介绍了如何在SpringBoot中使用Sse（Server-SentEvents）技术实现流式输出,后端使用SpringMVC中的SseEmitter对象,前端使用EventSource对象监听SSE接口并展示数据流,通过这种方式可以提升用户体验,避免大模型响应速度慢的问题,感兴趣的朋友一起看看吧
2025-03-03
springboot结合vue实现增删改查及分页查询
本文主要介绍了springboot结合vue实现增删改查及分页查询，文中通过示例代码介绍的非常详细,具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-09-09
Spring MVC 更灵活的控制 json 返回问题（自定义过滤字段）
本篇文章主要介绍了Spring MVC 更灵活的控制 json 返回问题（自定义过滤字段），具有一定的参考价值，感兴趣的小伙伴们可以参考一下。
2017-02-02