整合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
http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html#obtaining-offsets
添加相关Maven依赖：
代码语言：javascript复制<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>${spark.version}</version>
</dependency>注意
代码实现-自动提交偏移量到默认主题
代码语言：javascript复制package cn.itcast.streaming
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Author itcast
* Desc 使用spark-streaming-kafka-0-10版本中的Direct模式连接Kafka并自动提交偏移量
*/
object SparkStreaming_Kafka_01 {
def main(args: Array[String]): Unit = {
//1.准备SparkStreaming执行环境--StreamingContext
val 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重置位置
"auto.commit.interval.ms"->"1000",//自动提交的时间间隔
"enable.auto.commit" -> (true: 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.从kafkaDS中获取发送的value
val valuesDS: DStream[String] = kafkaDS.map(_.value)
//5.输出
valuesDS.print()
//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
}
}代码实现-手动提交偏移量到默认主题
代码语言：javascript复制package cn.itcast.streaming
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import 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执行环境--StreamingContext
val 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重置位置
//"auto.commit.interval.ms"->"1000",//自动提交的时间间隔
"enable.auto.commit" -> (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-扩展
代码语言：javascript复制package cn.itcast.streaming
import java.sql.{DriverManager, ResultSet}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import 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执行环境--StreamingContext
val 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重置位置
//"auto.commit.interval.ms"->"1000",//自动提交的时间间隔
"enable.auto.commit" -> (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
}
}
}
