use test mysql> create table hlw_offset( topic varchar(32), groupid varchar(50), partitions i..."/>
溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

生產(chǎn)SparkStreaming數(shù)據(jù)零丟失最佳實(shí)踐(含代碼)

發(fā)布時(shí)間：2020-07-12 12:43:57 來(lái)源：網(wǎng)絡(luò) 閱讀：5575 作者：Stitch_x 欄目：大數(shù)據(jù)

MySQL創(chuàng)建存儲(chǔ)offset的表格

mysql> use test
mysql> create table hlw_offset(
        topic varchar(32),
        groupid varchar(50),
        partitions int,
        fromoffset bigint,
        untiloffset bigint,
        primary key(topic,groupid,partitions)
        );

Maven依賴包

<scala.version>2.11.8</scala.version>
<spark.version>2.3.1</spark.version>
<scalikejdbc.version>2.5.0</scalikejdbc.version>
--------------------------------------------------
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>
<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>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.scalikejdbc/scalikejdbc -->
<dependency>
    <groupId>org.scalikejdbc</groupId>
    <artifactId>scalikejdbc_2.11</artifactId>
    <version>2.5.0</version>
</dependency>
<dependency>
    <groupId>org.scalikejdbc</groupId>
    <artifactId>scalikejdbc-config_2.11</artifactId>
    <version>2.5.0</version>
</dependency>
<dependency>
    <groupId>com.typesafe</groupId>
    <artifactId>config</artifactId>
    <version>1.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.5</version>
</dependency>

實(shí)現(xiàn)思路

1）StreamingContext
2）從kafka中獲取數(shù)據(jù)(從外部存儲(chǔ)獲取offset-->根據(jù)offset獲取kafka中的數(shù)據(jù))
3）根據(jù)業(yè)務(wù)進(jìn)行邏輯處理
4）將處理結(jié)果存到外部存儲(chǔ)中--保存offset
5）啟動(dòng)程序，等待程序結(jié)束

代碼實(shí)現(xiàn)

  1. SparkStreaming主體代碼如下

    import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scalikejdbc._
import scalikejdbc.config._
object JDBCOffsetApp {
 def main(args: Array[String]): Unit = {
   //創(chuàng)建SparkStreaming入口
   val conf = new SparkConf().setMaster("local[2]").setAppName("JDBCOffsetApp")
   val ssc = new StreamingContext(conf,Seconds(5))
   //kafka消費(fèi)主題
   val topics = ValueUtils.getStringValue("kafka.topics").split(",").toSet
   //kafka參數(shù)
   //這里應(yīng)用了自定義的ValueUtils工具類，來(lái)獲取application.conf里的參數(shù)，方便后期修改
   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數(shù)據(jù)庫(kù)中讀取offset信息
   //+------------+------------------+------------+------------+-------------+
   //| topic      | groupid          | partitions | fromoffset | untiloffset |
   //+------------+------------------+------------+------------+-------------+
   //MySQL表結(jié)構(gòu)如上，將“topic”，“partitions”，“untiloffset”列讀取出來(lái)
   //組成 fromOffsets: Map[TopicAndPartition, Long]，后面createDirectStream用到
   DBs.setup()
   val fromOffset = DB.readOnly( implicit session => {
     SQL("select * from hlw_offset").map(rs => {
       (TopicAndPartition(rs.string("topic"),rs.int("partitions")),rs.long("untiloffset"))
     }).list().apply()
   }).toMap
   //如果MySQL表中沒(méi)有offset信息，就從0開(kāi)始消費(fèi)；如果有，就從已經(jīng)存在的offset開(kāi)始消費(fèi)
     val messages = if (fromOffset.isEmpty) {
       println("從頭開(kāi)始消費(fèi)...")
       KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)
     } else {
       println("從已存在記錄開(kāi)始消費(fèi)...")
       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的數(shù)據(jù)量
         println("數(shù)據(jù)統(tǒng)計(jì)記錄為："+rdd.count())
         //官方案例給出的獲得rdd offset信息的方法，offsetRanges是由一系列offsetRange組成的數(shù)組
//          trait HasOffsetRanges {
//            def offsetRanges: Array[OffsetRange]
//          }
         val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
         offsetRanges.foreach(x => {
           //輸出每次消費(fèi)的主題，分區(qū)，開(kāi)始偏移量和結(jié)束偏移量
           println(s"---${x.topic},${x.partition},${x.fromOffset},${x.untilOffset}---")
          //將最新的偏移量信息保存到MySQL表中
           DB.autoCommit( implicit session => {
             SQL("replace into hlw_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()
 }
}

  2. 自定義的ValueUtils工具類如下

    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
   }
 }
}

  3. application.conf內(nèi)容如下

    metadata.broker.list = "192.168.137.251:9092"
auto.offset.reset = "smallest"
group.id = "hlw_offset_group"
kafka.topics = "hlw_offset"
serializer.class = "kafka.serializer.StringEncoder"
request.required.acks = "1"
# JDBC settings
db.default.driver = "com.mysql.jdbc.Driver"
db.default.url="jdbc:mysql://hadoop000:3306/test"
db.default.user="root"
db.default.password="123456"

  4. 自定義kafka producer

    import java.util.{Date, Properties}
import kafka.producer.{KeyedMessage, Producer, ProducerConfig}
object KafkaProducer {
 def main(args: Array[String]): Unit = {
   val properties = new Properties()
   properties.put("serializer.class",ValueUtils.getStringValue("serializer.class"))
   properties.put("metadata.broker.list",ValueUtils.getStringValue("metadata.broker.list"))
   properties.put("request.required.acks",ValueUtils.getStringValue("request.required.acks"))
   val producerConfig = new ProducerConfig(properties)
   val producer = new Producer[String,String](producerConfig)
   val topic = ValueUtils.getStringValue("kafka.topics")
   //每次產(chǎn)生100條數(shù)據(jù)
   var i = 0
   for (i <- 1 to 100) {
     val runtimes = new Date().toString
    val messages = new KeyedMessage[String, String](topic,i+"","hlw: "+runtimes)
     producer.send(messages)
   }
   println("數(shù)據(jù)發(fā)送完畢...")
 }
}

測(cè)試

  1. 啟動(dòng)kafka服務(wù)，并創(chuàng)建主題

    [hadoop@hadoop000 bin]$ ./kafka-server-start.sh -daemon /home/hadoop/app/kafka_2.11-0.10.0.1/config/server.properties
[hadoop@hadoop000 bin]$ ./kafka-topics.sh --list --zookeeper localhost:2181/kafka
[hadoop@hadoop000 bin]$ ./kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic hlw_offset

  2. 測(cè)試前查看MySQL中offset表，剛開(kāi)始是個(gè)空表

    mysql> select * from hlw_offset;
Empty set (0.00 sec)

  3. 通過(guò)kafka producer產(chǎn)生500條數(shù)據(jù)

  4. 啟動(dòng)SparkStreaming程序

    //控制臺(tái)輸出結(jié)果：
從頭開(kāi)始消費(fèi)...
數(shù)據(jù)統(tǒng)計(jì)記錄為：500
---hlw_offset,0,0,500---
查看MySQL表，offset記錄成功

mysql> select * from hlw_offset;
+------------+------------------+------------+------------+-------------+
| topic      | groupid          | partitions | fromoffset | untiloffset |
+------------+------------------+------------+------------+-------------+
| hlw_offset | hlw_offset_group |          0 |          0 |         500 |
+------------+------------------+------------+------------+-------------+

  1. 關(guān)閉SparkStreaming程序，再使用kafka producer生產(chǎn)300條數(shù)據(jù),再次啟動(dòng)spark程序（如果spark從500開(kāi)始消費(fèi)，說(shuō)明成功讀取了offset，做到了只讀取一次語(yǔ)義）

    //控制臺(tái)結(jié)果輸出：
從已存在記錄開(kāi)始消費(fèi)...
數(shù)據(jù)統(tǒng)計(jì)記錄為：300
---hlw_offset,0,500,800---

  2. 查看更新后的offset MySQL數(shù)據(jù)

    mysql> select * from hlw_offset;
+------------+------------------+------------+------------+-------------+
| topic      | groupid          | partitions | fromoffset | untiloffset |
+------------+------------------+------------+------------+-------------+
| hlw_offset | hlw_offset_group |          0 |        500 |         800 |
+------------+------------------+------------+------------+-------------+
向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手