SparkStreaming 手动维护kafka Offset到Mysql实例

官网详解地址
http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

手动提交offset，以保证数据不会丢失，尤其是在网络抖动严重的情况下，但是如果kafka挂掉重启后，可能会造成一些其他问题，
例如找不到保存的offset，这个具体问题再具体分析，先上代码。
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.{OffsetRange, _}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
*

• 使用Spark-Kafka-0-10版本整合,并手动提交偏移量,维护到MySQL中
  */

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

//1.创建StreamingContext val conf = new SparkConf().setAppName("wc").setMaster("local[*]") val sc = new SparkContext(conf) sc.setLogLevel("WARN") val ssc = new StreamingContext(sc,Seconds(5)) //准备连接Kafka的参数 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "server1:9092,server2:9092,server3:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "SparkKafkaTest", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean) 

val topics = Array("spark_kafka_test").toSet

val recordDStream: DStream[ConsumerRecord[String, String]] = if (offsetMap.size > 0) { //有记录offset println("MySQL中记录了offset,则从该offset处开始消费") KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, //位置策略,源码强烈推荐使用该策略,会让Spark的Executor和Kafka的Broker均匀对应 Subscribe[String, String](topics, kafkaParams, offsetMap)) //消费策略,源码强烈推荐使用该策略 } else { //没有记录offset println("没有记录offset,则直接连接,从latest开始消费") KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, //位置策略,源码强烈推荐使用该策略,会让Spark的Executor和Kafka的Broker均匀对应 Subscribe[String, String](topics, kafkaParams)) //消费策略,源码强烈推荐使用该策略 } recordDStream.foreachRDD { messages => if (messages.count() > 0) { //当前这一时间批次有数据 messages.foreachPartition { messageIter => messageIter.foreach { message => //println(message.toString()) } } val offsetRanges: Array[OffsetRange] = messages.asInstanceOf[HasOffsetRanges].offsetRanges for (o <- offsetRanges) { println(s"topic=${o.topic},partition=${o.partition},fromOffset=${o.fromOffset},untilOffset=${o.untilOffset}") } //手动提交offset,默认提交到Checkpoint中 //recordDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) //实际中偏移量可以提交到MySQL/Redis中 saveOffsetRanges("SparkKafkaTest", offsetRanges) } } 

ssc.start()
ssc.awaitTermination()
}

/**

• 从数据库读取偏移量
  */

def getOffsetMap(groupid: String, topic: String) = {

Class.forName("com.mysql.jdbc.Driver") val connection = DriverManager.getConnection("jdbc:mysql://172.31.98.108:3306/bj_pfdh?characterEncoding=UTF-8", "root", "iflytek@web") val sqlselect = connection.prepareStatement(""" select * from kafka_offset where groupid=? and topic =? """) sqlselect.setString(1, groupid) sqlselect.setString(2, topic) val rs: ResultSet = sqlselect.executeQuery() val offsetMap = mutable.Map[TopicPartition, Long]() while (rs.next()) { offsetMap += new TopicPartition(rs.getString("topic"), rs.getInt("partition")) -> rs.getLong("offset") } rs.close() sqlselect.close() connection.close() offsetMap 

}

/**

• 将偏移量保存到数据库
  */

def saveOffsetRanges(groupid: String, offsetRange: Array[OffsetRange]) = {

val connection = DriverManager.getConnection("jdbc:mysql://172.31.98.108:3306/bj_pfdh?characterEncoding=UTF-8", "root", "iflytek@web") //replace into表示之前有就替换,没有就插入 val select_ps = connection.prepareStatement(""" select count(*) as count from kafka_offset where `groupid`=? and `topic`=? and `partition`=? """) val update_ps = connection.prepareStatement(""" update kafka_offset set `offset`=? where `groupid`=? and `topic`=? and `partition`=? """) val insert_ps = connection.prepareStatement(""" INSERT INTO kafka_offset(`groupid`, `topic`, `partition`, `offset`) VALUE(?,?,?,?) """) for (o <- offsetRange) { select_ps.setString(1, groupid) select_ps.setString(2, o.topic) select_ps.setInt(3, o.partition) val select_resut = select_ps.executeQuery() // println(select_resut.)// .getInt("count")) while (select_resut.next()) { println(select_resut.getInt("count")) if (select_resut.getInt("count") > 0) { //update update_ps.setLong(1, o.untilOffset) update_ps.setString(2, groupid) update_ps.setString(3, o.topic) update_ps.setInt(4, o.partition) update_ps.executeUpdate() } else { //insert insert_ps.setString(1, groupid) insert_ps.setString(2, o.topic) insert_ps.setInt(3, o.partition) insert_ps.setLong(4, o.untilOffset) insert_ps.executeUpdate() } } } select_ps.close() update_ps.close() insert_ps.close() connection.close()

}

如果报错连不上数据库或连接数据库地址失败，请查看是否添加了mysql客户端jar包。

 --------五维空间s