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