版本信息:
spark:2.2.0
kakfa:0.10.1.0
scala:2.11.8
scalikejdbc:3.3.2
Pom文件:
<properties>
<scala.version>2.11.8</scala.version>
<spark.version>2.2.0</spark.version>
<scalikejdbc.version>3.3.2</scalikejdbc.version>
</properties>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
<!--scalikejdbc 依赖 -->
<dependency>
<groupId>org.scalikejdbc</groupId>
<artifactId>scalikejdbc_2.11</artifactId>
<version>${scalikejdbc.version}</version>
</dependency>
<dependency>
<groupId>org.scalikejdbc</groupId>
<artifactId>scalikejdbc-config_2.11</artifactId>
<version>${scalikejdbc.version}</version>
</dependency>
<!--Spark 依赖 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<!--mysql 依赖 -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.6</version>
</dependency>
</dependencies>
application.conf文件
db.default.driver="com.mysql.jdbc.Driver"
db.default.url="jdbc:mysql://hadoop000:3306/hadoop_train?characterEncoding=utf-8"
db.default.user="root"
db.default.password="root"
dataSourceClassName=com.mysql.jdbc.jdbc2.optional.MysqlDataSource
#Kafka信息
metadata.broker.list = "192.168.245.100:9092"
#从老数据开始消费
auto.offset.reset = "smallest"
group.id = "baidu_offset_group"
kafka.topics = "baidu"
serializer.class = "kafka.serializer.StringEncoder"
request.required.acks = "1"
ValueUtils
package com.soul.bigdata.spark.streaming01
import com.typesafe.config.ConfigFactory
import org.apache.commons.lang3.StringUtils
object ValueUtils {
val load = ConfigFactory.load()
def getStringValue(key: String, defaultValue: String = "") = {
val value = load.getString(key)
if (StringUtils.isNotEmpty(value)) {
value
} else {
defaultValue
}
}
}
MySQL Offset表
create table baidu_offset(
topic varchar(32),
groupid varchar(50),
partitions int,
fromoffset bigint,
untiloffset bigint,
primary key(topic,groupid,partitions)
);
代码:
package com.soul.bigdata.spark.streaming01
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scalikejdbc.{DB, SQL}
import scalikejdbc.config.DBs
object StreamingOffsetMySQL {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("StreamingOffsetMySQL")
val ssc = new StreamingContext(conf, Seconds(10))
//Topic
val topics = ValueUtils.getStringValue("kafka.topics").split(",").toSet
//kafka参数
//这里应用了自定义的ValueUtils工具类,来获取application.conf里的参数,方便后期修改
val kafkaParams = Map[String, String](
"metadata.broker.list" -> ValueUtils.getStringValue("metadata.broker.list"),
"auto.offset.reset" -> ValueUtils.getStringValue("auto.offset.reset"),
"group.id" -> ValueUtils.getStringValue("group.id")
)
//先使用scalikejdbc从MySQL数据库中读取offset信息
//+------------+------------------+------------+------------+-------------+
//| topic | groupid | partitions | fromoffset | untiloffset |
//+------------+------------------+------------+------------+-------------+
//MySQL表结构如上,将“topic”,“partitions”,“untiloffset”列读取出来
//组成 fromOffsets: Map[TopicAndPartition, Long],后面createDirectStream用到
DBs.setup()
val fromOffset = DB.readOnly(implicit session => {
SQL("select * from baidu_offset").map(rs => {
(TopicAndPartition(rs.string("topic"), rs.int("partitions")), rs.long("untiloffset"))
}).list().apply()
}).toMap
//如果MySQL表中没有offset信息,就从0开始消费;如果有,就从已经存在的offset开始消费
val messages = if (fromOffset.isEmpty) {
println("从头开始消费...")
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
} else {
println("从已存在记录开始消费...")
val messageHandler = (mm: MessageAndMetadata[String, String]) => (mm.key(), mm.message())
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffset, messageHandler)
}
messages.foreachRDD(rdd => {
if (!rdd.isEmpty()) {
//输出rdd的数据量
println("数据统计记录为:" + rdd.count())
//官方案例给出的获得rdd offset信息的方法,offsetRanges是由一系列offsetRange组成的数组
// trait HasOffsetRanges {
// def offsetRanges: Array[OffsetRange]
// }
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
offsetRanges.foreach(x => {
//输出每次消费的主题,分区,开始偏移量和结束偏移量
println(s"---${x.topic},${x.partition},${x.fromOffset},${x.untilOffset}---")
//将最新的偏移量信息保存到MySQL表中
DB.autoCommit(implicit session => {
SQL("replace into baidu_offset(topic,groupid,partitions,fromoffset,untiloffset) values (?,?,?,?,?)")
.bind(x.topic, ValueUtils.getStringValue("group.id"), x.partition, x.fromOffset, x.untilOffset)
.update().apply()
})
})
}
})
ssc.start()
ssc.awaitTermination()
}
}
运行
image.png
停掉程序,重新运行,开始offset是从411开始消费的就达到了我们的目的
image.png image.png image.png
网友评论