spark streaming窗口及聚合操作后怎么管理offset,針對(duì)這個(gè)問(wèn)題,這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答,希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。
創(chuàng)新互聯(lián)公司專注于隆林網(wǎng)站建設(shè)服務(wù)及定制,我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。 熱誠(chéng)為您提供隆林營(yíng)銷型網(wǎng)站建設(shè),隆林網(wǎng)站制作、隆林網(wǎng)頁(yè)設(shè)計(jì)、隆林網(wǎng)站官網(wǎng)定制、小程序開(kāi)發(fā)服務(wù),打造隆林網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供隆林網(wǎng)站排名全網(wǎng)營(yíng)銷落地服務(wù)。
spark streaming經(jīng)過(guò)窗口的集合操作之后,再去管理offset呢?
對(duì)于spark streaming來(lái)說(shuō)窗口操作之后,是無(wú)法管理offset的,因?yàn)閛ffset的存儲(chǔ)于HasOffsetRanges,只有kafkaRDD繼承了該特質(zhì),經(jīng)過(guò)轉(zhuǎn)化的其他RDD都不支持了。所以無(wú)法通過(guò)其他RDD轉(zhuǎn)化為HasOffsetRanges來(lái)獲取offset,以便自己管理。
kafkaRDD的繼承關(guān)系如下:
private[spark] class KafkaRDD[K, V]( sc: SparkContext, val kafkaParams: ju.Map[String, Object], val offsetRanges: Array[OffsetRange], val preferredHosts: ju.Map[TopicPartition, String], useConsumerCache: Boolean) extends RDD[ConsumerRecord[K, V]](sc, Nil) with Logging with HasOffsetRanges {
HasOffsetRanges只有kafkaRDD繼承了他,所以假如我們對(duì)KafkaRDD進(jìn)行了轉(zhuǎn)化之后就無(wú)法再獲取offset了。
HasOffsetRanges就是一個(gè)OffsetRange的數(shù)組:
trait HasOffsetRanges { def offsetRanges: Array[OffsetRange]}
?
再看一下,OffsetRange的實(shí)現(xiàn):
窗口操作會(huì)包含若干批次的RDD數(shù)據(jù),窗口操作也往往帶有聚合操作,所以KafkaRDD肯定會(huì)被轉(zhuǎn)化為其他類型的RDD的,那么之后就無(wú)法轉(zhuǎn)化為hasoffsetranges了,也是管理offset變得很麻煩的。
實(shí)際上,無(wú)論是窗口是否有重疊和包含聚合,其實(shí)我們只關(guān)心本次處理窗口的kafkardds 的offset范圍[fromOffset, toOffset),由于fromOffset是上次提交成功的,那么本次處理完只需要提交的toOffset即可,即使處理失敗也可以從fromOffset開(kāi)始重新處理。也就實(shí)現(xiàn)了數(shù)據(jù)的最少一次處理,假如能與結(jié)果一起管理,也可以實(shí)現(xiàn)僅一次處理。那么提交offset我們只需要提交最近的那個(gè)批次的kafkaRDD的toOffset即可。
那么如何獲取最新的kafkaRDD的toOffset呢?
其實(shí),我們只需要在driver端記錄kafkardd轉(zhuǎn)化的hasoffsetrange存儲(chǔ)的offset即可。
回顧一下,對(duì)于spark 來(lái)說(shuō)代碼執(zhí)行位置分為driver和executor,我們希望再driver端獲取到offset,等處理完結(jié)果后,再提交offset到kafka或者直接與結(jié)果一起管理offset。
那么窗口操作之前獲取offset方法是什么呢?
就是利用transform操作,完成下面的步驟:
var A:mutable.HashMap[String,Array[OffsetRange]] = new mutable.HashMap()
val offsetRanges = r.asInstanceOf[HasOffsetRanges].offsetRanges
A += ("rdd1"->offsetRanges)
上述步驟就完成了,只記錄最新kafkardd的hasoffsetranges里存儲(chǔ)的offset功能。
總結(jié)一下:driver端通過(guò)使用transform獲取到offset信息,然后在輸出操作foreachrdd里面完成offset的提交操作。
package bigdata.spark.SparkStreaming.kafka010
import java.util.Properties
import org.apache.kafka.clients.consumer.{Consumer, ConsumerRecord, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, TaskContext}
import scala.collection.JavaConverters._
import scala.collection.mutable
object kafka010NamedRDD {
def main(args: Array[String]) {
// 創(chuàng)建一個(gè)批處理時(shí)間是2s的context 要增加環(huán)境變量
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
ssc.checkpoint("/opt/checkpoint")
// 使用broker和topic創(chuàng)建DirectStream
val topicsSet = "test".split(",").toSet
val kafkaParams = Map[String, Object]("bootstrap.servers" -> "mt-mdh.local:9093",
"key.deserializer"->classOf[StringDeserializer],
"value.deserializer"-> classOf[StringDeserializer],
"group.id"->"test4",
"auto.offset.reset" -> "latest",
"enable.auto.commit"->(false: java.lang.Boolean))
// 沒(méi)有接口提供 offset
val messages = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams,getLastOffsets(kafkaParams ,topicsSet)))//
var A:mutable.HashMap[String,Array[OffsetRange]] = new mutable.HashMap()
val trans = messages.transform(r =>{
val offsetRanges = r.asInstanceOf[HasOffsetRanges].offsetRanges
A += ("rdd1"->offsetRanges)
r
}).countByWindow(Seconds(10), Seconds(5))
trans.foreachRDD(rdd=>{
if(!rdd.isEmpty()){
val offsetRanges = A.get("rdd1").get//.asInstanceOf[HasOffsetRanges].offsetRanges
rdd.foreachPartition { iter =>
val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
}
println(rdd.count())
println(offsetRanges)
// 手動(dòng)提交offset ,前提是禁止自動(dòng)提交
messages.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
// A.-("rdd1")
})
// 啟動(dòng)流
ssc.start()
ssc.awaitTermination()
}
def getLastOffsets(kafkaParams : Map[String, Object],topics:Set[String]): Map[TopicPartition, Long] ={
val props = new Properties()
props.putAll(kafkaParams.asJava)
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(topics.asJavaCollection)
paranoidPoll(consumer)
val map = consumer.assignment().asScala.map { tp =>
println(tp+"---" +consumer.position(tp))
tp -> (consumer.position(tp))
}.toMap
println(map)
consumer.close()
map
}
def paranoidPoll(c: Consumer[String, String]): Unit = {
val msgs = c.poll(0)
if (!msgs.isEmpty) {
// position should be minimum offset per topicpartition
msgs.asScala.foldLeft(Map[TopicPartition, Long]()) { (acc, m) =>
val tp = new TopicPartition(m.topic, m.partition)
val off = acc.get(tp).map(o => Math.min(o, m.offset)).getOrElse(m.offset)
acc + (tp -> off)
}.foreach { case (tp, off) =>
c.seek(tp, off)
}
}
}
}
關(guān)于spark streaming窗口及聚合操作后怎么管理offset問(wèn)題的解答就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,如果你還有很多疑惑沒(méi)有解開(kāi),可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識(shí)。
文章標(biāo)題:sparkstreaming窗口及聚合操作后怎么管理offset
鏈接分享:http://bm7419.com/article22/pcidjc.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供用戶體驗(yàn)、品牌網(wǎng)站制作、網(wǎng)站策劃、微信小程序、面包屑導(dǎo)航、搜索引擎優(yōu)化
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來(lái)源: 創(chuàng)新互聯(lián)