1 Spark-streaming
1.1Spark-streaming 的KafkaManager类
1.1.1KafkaManager主要封装了两个方法
程序启动的时候获取consumer-group在zk保存的offset以及当前topic拥有的最大以及最小offset，获取正确的offset，代码如下
private def setOrUpdateOffsets(topics: Set[String], groupId: String): Unit = {
topics.foreach(topic => {
var hasConsumed = true
val partitionsE = kc.getPartitions(Set(topic))
if (partitionsE.isLeft) throw new SparkException(s"get kafka partition failed: ${partitionsE.left.get.mkString("\n")}")
val partitions: Set[TopicAndPartition] = partitionsE.right.get
val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)
if (consumerOffsetsE.isLeft) hasConsumed = false
log.info("consumerOffsetsE.isLeft: " + consumerOffsetsE.isLeft)
if (hasConsumed) {
// 消费过
log.info("消费过")
/**
* 如果zk上保存的offsets已经过时了，即kafka的定时清理策略已经将包含该offsets的文件删除。
* 针对这种情况，只要判断一下zk上的consumerOffsets和earliestLeaderOffsets的大小，
* 如果consumerOffsets比earliestLeaderOffsets还小的话，说明consumerOffsets已过时,
* 这时把consumerOffsets更新为earliestLeaderOffsets
*/
val earliestLeaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)
if (earliestLeaderOffsetsE.isLeft) throw new SparkException(s"get earliest offsets failed: ${earliestLeaderOffsetsE.left.get.mkString("\n")}")
val earliestLeaderOffsets = earliestLeaderOffsetsE.right.get
val consumerOffsets = consumerOffsetsE.right.get
// 可能只是存在部分分区consumerOffsets过时，所以只更新过时分区的consumerOffsets为earliestLeaderOffsets
var offsets: Map[TopicAndPartition, Long] = Map()
consumerOffsets.foreach({
case (tp, n) =>
val earliestLeaderOffset = earliestLeaderOffsets(tp).offset
if (n < earliestLeaderOffset) {
log.info("consumer group:" + groupId + ",topic:" + tp.topic + ",partition:" + tp.partition +
" offsets已经过时，更新为" + earliestLeaderOffset)
offsets += (tp -> earliestLeaderOffset)
}
})
log.info("offsets: " + consumerOffsets)
if (!offsets.isEmpty) {
kc.setConsumerOffsets(groupId, offsets)
}
} else {
// 没有消费过
log.info("没消费过")
val reset = kafkaParams.get("auto.offset.reset").map(_.toLowerCase)
var leaderOffsets: Map[TopicAndPartition, LeaderOffset] = null
if (reset == Some("smallest")) {
leaderOffsets = kc.getEarliestLeaderOffsets(partitions).right.get
} else {
leaderOffsets = kc.getLatestLeaderOffsets(partitions).right.get
}
val offsets = leaderOffsets.map {
case (tp, offset) => (tp, offset.offset)
}
log.info("offsets: " + offsets)
kc.setConsumerOffsets(groupId, offsets)
}
})
}
当处理完DStream后，手动更新zk上的offset，如果程序出异常，则不更新offset，做到at-least-once,代码如下
override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
/**
* 在这更新 currentOffsets 从而做到自适应上游 partition 数目变化
*/
updateCurrentOffsetForKafkaPartitionChange()
super.compute(validTime)
}
def updateZKOffsets(rdd: RDD[(String,String)]): Unit = {
val groupId = kafkaParams.get("group.id").get
val offsetsList = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
for (offsets <- offsetsList) {
val topicAndPartition = TopicAndPartition(offsets.topic, offsets.partition)
val o = kc.setConsumerOffsets(groupId, Map((topicAndPartition, offsets.untilOffset)))
log.info(s"${DateTimeUtils.getStringDateTime()} @@@@@@ topic ${offsets.topic} partition ${offsets.partition} fromoffset ${offsets.fromOffset} untiloffset ${offsets.untilOffset}" +
s" diff ${{offsets.fromOffset} - {offsets.untilOffset} } #######")
if (o.isLeft) {
log.error(s"Error updating the offset to Kafka cluster: ${o.left.get}")
}
}
}
1.2 Spark-streaming 的自适应partition变化类
增加AdaptiveUpStreamDirectKafkaInputDStream自适应partition增加功能类，比如topic增加partition，spark-streaming自适应partition变化，关键代码如下
private def updateCurrentOffsetForKafkaPartitionChange(): Unit = {
val topic = currentOffsets.head._1.topic
val nextPartitions: Int = getTopicMeta(topic) match {
case Some(x) => x.partitionsMetadata.size()
case _ => 0
}
val currPartitions = currentOffsets.keySet.size
if (nextPartitions > currPartitions
) {
var i = currPartitions
while (i < nextPartitions
) {
currentOffsets = currentOffsets + (TopicAndPartition(topic, i) -> 0
)
i = i + 1
}
}
logInfo(s"######### ${
nextPartitions
} currentParttions ${
currentOffsets.keySet.size
} ########")
}
private def getTopicMeta(topic: String): Option[TopicMetadata] = {
var metaData: Option[TopicMetadata] = None
var consumer: Option[SimpleConsumer] = None
val topics = List[String](topic)
val brokerList = MTkafkaParams.get("metadata.broker.list").get.split(",")
//val brokerList = kafkaBrokerList.split(",")
brokerList.foreach(
item => {
val hostPort = item.split(":")
try {
breakable {
for (i <- 0 to 3) {
consumer = Some(new SimpleConsumer(host = hostPort(0), port = hostPort(1).toInt,
soTimeout = 10000, bufferSize = 64 * 1024, clientId = "leaderLookup"))
val req: TopicMetadataRequest = new TopicMetadataRequest(topics.asJava)
val resp = consumer.get.send(req)
metaData = Some(resp.topicsMetadata.get(0))
if (metaData.get.errorCode == ErrorMapping.NoError) break()
}
}
} catch {
case e => ;
logInfo(s" ###### Error in AdaptiveUpStreamDirectKafkaInputDStream ${
e
} ######")
}
}
)
metaData
}
1.3Spark-streaming 的编程模板代码
object TestMain {
@transient lazy val LOG: Logger = LoggerFactory.getLogger(this.getClass);
def main(args: Array[String]): Unit = {
var conf_file = ""
if (args.length > 0) {
conf_file = args(0)
}
//init conf_file
SparkUtils.initialize(conf_file)
//init ssc
val sparkConf = SparkUtils.buildSparkConf()
val ssc = new StreamingContext(sparkConf, Seconds(SparkUtils.getValueByKey(Constants.SPARK_APP_DURATION).toLong))
//init kafka params
val kafkaParams = SparkUtils.buildkafkaParams()
val topics = SparkUtils.getValueByKey(Constants.KAFKA_TOPICS)
//init DStream
val km = new KafkaManager(kafkaParams)
val kafkaDirectStream = km.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics.split(",").toSet)
LOG.info("======Initial Done======")
kafkaDirectStream.cache()
// init hbase config to broadcast to cluster executors
val hbaseConfig = ssc.sparkContext.broadcast(SparkUtils.buildHbaseProps())
//transform
val processedDStream = kafkaDirectStream.map(_._2)
//to do something
//action
processedDStream.foreachRDD { rdd =>
//LOG.info("### rdd count = " + rdd.count() + "### rdd partition count:" + rdd.partitions.length)
rdd.foreachPartition { partitionOfRecords =>
//to do something
}
}
//update zk offsets
kafkaDirectStream.foreachRDD(rdd => {
if (!rdd.isEmpty)
km.updateZKOffsets(rdd)
})
ssc.start()
ssc.awaitTermination()
}
}
1.4Spark-streaming 的工具类
jedis pool
/**
* Created by lancerlin on 2018/1/3.
*/
class MyRedisConnectionPool(createMyRedisConnnectionPoolFunc: () => RedisConnectionPool) extends Serializable {
@transient lazy val LOG: Logger = LoggerFactory.getLogger(this.getClass)
lazy val pool = createMyRedisConnnectionPoolFunc()
}
object MyRedisConnectionPool {
lazy val config = {
val cf = new PoolConfig
cf.setMaxTotal(100)
cf.setMaxIdle(5)
cf.setMaxWaitMillis(1000)
cf.setTestOnBorrow(true)
cf
}
def apply[K, V](props: java.util.Properties): MyRedisConnectionPool = {
val createMyRedisConnnectionPoolFunc = () => {
val pool = new RedisConnectionPool(config, props)
sys.addShutdownHook {
pool.close()
}
pool
}
new MyRedisConnectionPool(createMyRedisConnnectionPoolFunc)
}
}
streaming代码中的用法
//broadcast redis info
val redisConfig = ssc.sparkContext.broadcast(SparkUtils.buildRedisProps())
.....
dstream.foreachRDD(rdd => {
rdd.foreachPartition(partitionRecords => {
val pool = MyRedisConnectionPool(redisConfig.value).pool
val proxy = pool.getConnection
while (partitionRecords.hasNext) {
proxy.doMethod()
}
pool.returnConnection(proxy)
})
})
同理kafkaproducer
class MySparkKafkaProducer[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
/* This is the key idea that allows us to work around running into
NotSerializableExceptions. */
lazy val producer = createProducer()
def send(topic: String, key: K, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, key, value))
def send(topic: String, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, value))
}
object MySparkKafkaProducer {
import scala.collection.JavaConversions._
def apply[K, V](config: Map[String, Object]): MySparkKafkaProducer[K, V] = {
val createProducerFunc = () => {
val producer = new KafkaProducer[K, V](config)
sys.addShutdownHook {
// Ensure that, on executor JVM shutdown, the Kafka producer sends
// any buffered messages to Kafka before shutting down.
producer.close()
}
producer
}
new MySparkKafkaProducer(createProducerFunc)
}
def apply[K, V](config: java.util.Properties): MySparkKafkaProducer[K, V] = apply(config.toMap)
}
同理mysqlUitls
/**
* Created by lancerlin on 2017/12/28.
*/
class MysqlPool(createMysqlPoolFunc:() => JdbcConnectionPool) extends Serializable {
println("init ....")
lazy val pool = createMysqlPoolFunc()
}
object MysqlPool {
lazy val config = {
val cf = new PoolConfig
cf.setMaxTotal(100)
cf.setMaxIdle(5)
cf.setMaxWaitMillis(1000)
cf.setTestOnBorrow(true)
cf
}
def apply[K, V](props: java.util.Properties): MysqlPool = {
val createMysqlPoolFunc = () => {
val pool = new JdbcConnectionPool(config, props)
sys.addShutdownHook {
pool.close()
}
pool
}
new MysqlPool(createMysqlPoolFunc)
}
}
1.5 Spark-streaming 中如何更新broadcast
import java.io.{ObjectInputStream, ObjectOutputStream}
import java.util.Calendar
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.StreamingContext
import org.slf4j.{Logger, LoggerFactory}
import scala.reflect.ClassTag
/**
* Created by lancerlin on 2018/2/1.
*
* 更新broadcast工具类，使用方式
* *
* val value1 = MyBroadcastWrapper(ssc, new Date().toString)
* *
*processedDStream.foreachRDD { rdd =>
* *
* value1.update(new Date().toString, true)
* *
*rdd.foreachPartition {
* partitionOfRecords =>
* while (partitionOfRecords.hasNext){
* val str = partitionOfRecords.next()
* }
*LOG.warn(s"borad cast value @@@@ [{${value1.value}}])")
* }
* }
*
*/
// This wrapper lets us update brodcast variables within DStreams' foreachRDD
// without running into serialization issues
case class MyBroadcastWrapper[T: ClassTag](
@transient private val ssc: StreamingContext,
@transient private val _v: T
) {
var lastUpdatedAt = Calendar.getInstance.getTime
@transient lazy val LOG: Logger = LoggerFactory.getLogger(this.getClass);
@transient private var v = ssc.sparkContext.broadcast(_v)
def update(newValue: T, blocking: Boolean = false, updateInterval: Integer): Unit = {
val currentDate = Calendar.getInstance.getTime
val diff = currentDate.getTime - lastUpdatedAt.getTime
//默认60秒更新，这个时间跟进需求修改
var interval = 60000
if (updateInterval != null && updateInterval >= 0) {
interval = updateInterval
}
if (v == null || diff > interval) {
lastUpdatedAt = Calendar.getInstance.getTime
if (v != null) {
v.unpersist(blocking)
}
v = ssc.sparkContext.broadcast(newValue)
}
}
def value: T = v.value
private def writeObject(out: ObjectOutputStream): Unit = {
out.writeObject(v)
}
private def readObject(in: ObjectInputStream): Unit = {
v = in.readObject().asInstanceOf[Broadcast[T]]
}
}
用法
val value = MyBroadcastWrapper(ssc, new Date().toString)
processedDStream.foreachRDD { rdd =>
//默认一分钟更新一次，传入120000，修改为两分钟更新一次
value1.update(new Date().toString, true,120000)
rdd.foreachPartition {
partitionOfRecords =>
while (partitionOfRecords.hasNext){
val str = partitionOfRecords.next()
}
LOG.warn(s"borad cast value @@@@ [{${value.value}}])")
}
}
测试结果，10秒一个批，两分钟更新日志的值LOG.warn(s"borad cast value @@@@ [{${value.value}}])")
image.png
参考文档：
https://stackoverflow.com/questions/33372264/how-can-i-update-a-broadcast-variable-in-spark-streaming
https://www.jianshu.com/p/5dbb102cbbd9
