700字范文 > 大数据Spark（四十三）：SparkStreaming整合Kafka 0.10 开发使用

大数据Spark（四十三）：SparkStreaming整合Kafka 0.10 开发使用

时间：2022-06-10 12:38:30

整合Kafka 0-10-开发使用

原理

1.Direct方式

2.简单的并行度1:1

API

注意

代码实现-自动提交偏移量到默认主题

代码实现-手动提交偏移量到默认主题

代码实现-手动提交偏移量到MySQL-扩展

整合Kafka 0-10-开发使用

原理

目前企业中基本都使用New Consumer API集成，优势如下：

1.Direct方式

直接到Kafka Topic中依据偏移量范围获取数据，进行处理分析；

The Spark Streaming integration for Kafka 0.10 is similar in design to the 0.8 Direct Stream approach；

2.简单的并行度1:1

每批次中RDD的分区与Topic分区一对一关系；

It provides simple parallelism, 1:1 correspondence between Kafka partitions and Spark partitions, and access to offsets and metadata；

获取Topic中数据的同时，还可以获取偏移量和元数据信息；

采用Direct方式消费数据时，可以设置每批次处理数据的最大量，防止【波峰】时数据太多，导致批次数据处理有性能问题：

参数：spark.streaming.kafka.maxRatePerPartition含义：Topic中每个分区每秒中消费数据的最大值举例说明：BatchInterval：5s、Topic-Partition：3、maxRatePerPartition： 10000最大消费数据量：10000 * 3 * 5 = 150000 条

API

/docs/latest/streaming-kafka-0-10-integration.html#obtaining-offsets

添加相关Maven依赖：

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.11</artifactId><version>${spark.version}</version></dependency>

注意

代码实现-自动提交偏移量到默认主题

代码实现-手动提交偏移量到默认主题

package cn.itcast.streamingimport org.apache.kafka.clients.consumer.ConsumerRecordimport org.mon.serialization.StringDeserializerimport org.apache.spark.streaming.dstream.{DStream, InputDStream}import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}import org.apache.spark.streaming.{Seconds, StreamingContext}import org.apache.spark.{SparkConf, SparkContext}/*** Author itcast* Desc 使用spark-streaming-kafka-0-10版本中的Direct模式连接Kafka并手动提交偏移量*/object SparkStreaming_Kafka_02 {def main(args: Array[String]): Unit = {//1.准备SparkStreaming执行环境--StreamingContextval conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("WARN")val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))//2.准备Kafka的连接参数,如集群地址,主题,消费者组名称,是否自动提交,offset重置位置,kv序列化val kafkaParams = Map[String, Object]("bootstrap.servers" -> "node1:9092,node2:9092,node3:9092",//集群地址"key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则"value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则"group.id" -> "spark",//消费者组名称//earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费//latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费//none:表示如果有offset记录从offset记录开始消费,如果没有就报错"auto.offset.reset" -> "latest",//offset重置位置//"mit.interval.ms"->"1000",//自动提交的时间间隔"mit" -> (false: java.lang.Boolean)//是否自动提交偏移量)val topics = Array("spark_kafka")//要消费哪个主题//3.使用spark-streaming-kafka-0-10中的Direct模式连接Kafka// ssc: StreamingContext,// locationStrategy: LocationStrategy,位置策略,直接使用源码推荐的优先一致性策略即可,在大多数情况下，它将一致地在所有执行器之间分配分区// consumerStrategy: ConsumerStrategy[K, V],消费策略,直接使用源码推荐的订阅模式,通过参数订阅主题即可//kafkaDS就是从Kafka中消费到的完整的消息记录!val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))//4.处理从Kafka中消费到的value//手动提交偏移量的时机://1.每隔一段时间提交一次:可以,但是和自动提交一样了,那还不如直接自动提交!//2.消费一条消息就提交一次offset:可以但是提交的太频繁了,可能会影响效率!除非对数据安全要求特别高!//3.消费一小批消息就提交一次offset:可以!一小批数据在SparkStreaming里面就是DStream底层的RDD(微批)!kafkaDS.foreachRDD(rdd=>{//该如何消费/处理就如何消费/处理//完事之后就应该提交该批次的offset!if(!rdd.isEmpty()){//当前批次的rdd不为空,那么就消费该批次数据并提交偏移量rdd.foreach(r=>{println(s"消费到的消息记录的分区为:${r.partition()},offset为:${r.offset()},key为:${r.key()},value为:${r.value()}")})//代码走到这里说明该批次数据已经消费并处理了,那么应该手动提交偏移量了!//要手动提交的偏移量信息都在rdd中,但是我们要提交的仅仅是offset相关的信息,所以将rdd转为方便我们提交的Array[OffsetRange]类型val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges//上面的offsetRanges数组中就记录了各个分区的偏移量信息!offsetRanges.foreach(o=>{println(s"offsetRanges中记录的分区为:${o.partition},开始offset为:${o.fromOffset},结束offset为${o.untilOffset}")})//手动提交--提交到Kafka的默认主题中!(注:如果设置了Checkpoint,还会储存一份到Checkpoint中)kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)println("当前批次的offset已经提交到默认主题中")}})//5.输出//6.启动并等待结束ssc.start()ssc.awaitTermination()ssc.stop(stopSparkContext = true, stopGracefully = true)//注意://1.启动kafka//2.准备主题:/export/server/kafka/bin/kafka-topics.sh --create --zookeeper node1:2181 --replication-factor 1--partitions 3 --topic spark_kafka//3.开启控制台生产者:/export/server/kafka/bin/kafka-console-producer.sh --broker-list node1:9092 --topic spark_kafka}}

代码实现-手动提交偏移量到MySQL-扩展

package cn.itcast.streamingimport java.sql.{DriverManager, ResultSet}import org.apache.kafka.clients.consumer.ConsumerRecordimport org.mon.TopicPartitionimport org.mon.serialization.StringDeserializerimport org.apache.spark.streaming.dstream.InputDStreamimport org.apache.spark.streaming.kafka010._import org.apache.spark.streaming.{Seconds, StreamingContext}import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable/*** Author itcast* Desc 使用spark-streaming-kafka-0-10版本中的Direct模式连接Kafka并手动提交偏移量到MySQL*/object SparkStreaming_Kafka_03 {def main(args: Array[String]): Unit = {//1.准备SparkStreaming执行环境--StreamingContextval conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("WARN")val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))//2.准备Kafka的连接参数,如集群地址,主题,消费者组名称,是否自动提交,offset重置位置,kv序列化val kafkaParams = Map[String, Object]("bootstrap.servers" -> "node1:9092,node2:9092,node3:9092", //集群地址"key.deserializer" -> classOf[StringDeserializer], //key的反序列化规则"value.deserializer" -> classOf[StringDeserializer], //value的反序列化规则"group.id" -> "spark", //消费者组名称//earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费//latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费//none:表示如果有offset记录从offset记录开始消费,如果没有就报错"auto.offset.reset" -> "latest", //offset重置位置//"mit.interval.ms"->"1000",//自动提交的时间间隔"mit" -> (false: java.lang.Boolean) //是否自动提交偏移量)val topics = Array("spark_kafka") //要消费哪个主题//3.使用spark-streaming-kafka-0-10中的Direct模式连接Kafka//连接kafka之前,要先去MySQL看下有没有该消费者组的offset记录,如果有从记录的位置开始消费,如果没有从"auto.offset.reset" -> "latest"位置开始消费!//Map[主题分区为key, offset为value]val offsetMap: mutable.Map[TopicPartition, Long] = OffsetUtil.getOffsetMap("spark", "spark_kafka")val kafkaDS: InputDStream[ConsumerRecord[String, String]] = if (offsetMap.size > 0) {println("MySQL中有记录该消费者消费该主题的各个分区的offset信息,所以接着该记录开始消费")KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsetMap))} else {println("MySQL没有记录该消费者消费该主题的各个分区的offset信息,所以从auto.offset.reset配置的latest开始消费")KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))}//4.处理从Kafka中消费到的value//手动提交偏移量的时机://1.每隔一段时间提交一次:可以,但是和自动提交一样了,那还不如直接自动提交!//2.消费一条消息就提交一次offset:可以但是提交的太频繁了,可能会影响效率!除非对数据安全要求特别高!//3.消费一小批消息就提交一次offset:可以!一小批数据在SparkStreaming里面就是DStream底层的RDD(微批)!kafkaDS.foreachRDD(rdd => {//该如何消费/处理就如何消费/处理//完事之后就应该提交该批次的offset!if (!rdd.isEmpty()) { //当前批次的rdd不为空,那么就消费该批次数据并提交偏移量rdd.foreach(r => {println(s"消费到的消息记录的分区为:${r.partition()},offset为:${r.offset()},key为:${r.key()},value为:${r.value()}")})//代码走到这里说明该批次数据已经消费并处理了,那么应该手动提交偏移量了!//要手动提交的偏移量信息都在rdd中,但是我们要提交的仅仅是offset相关的信息,所以将rdd转为方便我们提交的Array[OffsetRange]类型val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges//上面的offsetRanges数组中就记录了各个分区的偏移量信息!offsetRanges.foreach(o => {println(s"offsetRanges中记录的分区为:${o.partition},开始offset为:${o.fromOffset},结束offset为${o.untilOffset}")})//手动提交--提交到Kafka的默认主题中!(注:如果设置了Checkpoint,还会储存一份到Checkpoint中)//kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)OffsetUtil.saveOffsetRanges("spark", offsetRanges)println("当前批次的offset已经提交到MySQL中")}})//5.输出//6.启动并等待结束ssc.start()ssc.awaitTermination()ssc.stop(stopSparkContext = true, stopGracefully = true)//注意://1.启动kafka//2.准备主题:/export/server/kafka/bin/kafka-topics.sh --create --zookeeper node1:2181 --replication-factor 1--partitions 3 --topic spark_kafka//3.开启控制台生产者:/export/server/kafka/bin/kafka-console-producer.sh --broker-list node1:9092 --topic spark_kafka}/*手动维护offset的工具类首先在MySQL创建如下表CREATE TABLE `t_offset` (`topic` varchar(255) NOT NULL,`partition` int(11) NOT NULL,`groupid` varchar(255) NOT NULL,`offset` bigint(20) DEFAULT NULL,PRIMARY KEY (`topic`,`partition`,`groupid`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;*/object OffsetUtil {//1.将偏移量保存到数据库def saveOffsetRanges(groupid: String, offsetRange: Array[OffsetRange]) = {val connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")//replace into表示之前有就替换,没有就插入val ps = connection.prepareStatement("replace into t_offset (`topic`, `partition`, `groupid`, `offset`) values(?,?,?,?)")for (o <- offsetRange) {ps.setString(1, o.topic)ps.setInt(2, o.partition)ps.setString(3, groupid)ps.setLong(4, o.untilOffset)ps.executeUpdate()}ps.close()connection.close()}//2.从数据库读取偏移量def getOffsetMap(groupid: String, topic: String) = {val connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")val ps = connection.prepareStatement("select * from t_offset where groupid=? and topic=?")ps.setString(1, groupid)ps.setString(2, topic)val rs: ResultSet = ps.executeQuery()val offsetMap = mutable.Map[TopicPartition, Long]()while (rs.next()) {offsetMap += new TopicPartition(rs.getString("topic"), rs.getInt("partition")) -> rs.getLong("offset")}rs.close()ps.close()connection.close()offsetMap}}}

本内容不代表本网观点和政治立场，如有侵犯你的权益请联系我们处理。

网友评论

网友评论仅供其表达个人看法，并不表明网站立场。