業(yè)務(wù)背景：使用Spark 、streaming從kafka讀取數(shù)據(jù)后寫入HBase。kafkaDStream是從kafka讀到的一個批次的數(shù)據(jù)流。

遍歷直接寫入HBase

最最基礎(chǔ)寫法是直接遍歷并一條一條寫入hbase。
第一版的核心代碼如下：

kafkaDStream.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        println("kafkaRDD get some data.")
        rdd.foreachPartition(partitionRecords => {
          // 獲取HBase連接
          val hbaseConnection: Connection = getHBaseConn()
          partitionRecords.foreach(line => {
            // 連接HBase表
            val tableName: TableName = TableName.valueOf(ConfigLoader.getString("hbase.table.name"))
            val table: Table = hbaseConnection.getTable(tableName)
            // 將kafka的每一條消息解析為JSON格式數(shù)據(jù)
            val jsonObj: Option[Any] = JSON.parseFull(line.value())
            val uuid: String = UUID.randomUUID().toString
            // println(line.value())
            val data: Map[String, Any] = jsonObj.get.asInstanceOf[Map[String, Any]]
            val a: String = data("a").asInstanceOf[String]
            val b: String = data("b").asInstanceOf[String]
            val c: String = data("c").asInstanceOf[String]

            val put = new Put(Bytes.toBytes(uuid))
            val tableColumnFamily = ConfigLoader.getString("hbase.table.column.family")
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("a"), Bytes.toBytes(a))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("b"), Bytes.toBytes(b))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("c"), Bytes.toBytes(c))

            // 將數(shù)據(jù)寫入HBase，若出錯關(guān)閉table
            Try(table.put(put)).getOrElse(table.close())
          })
          hbaseConnection.close()
        })
      } else {
        println("kafkaRDD is Empty!!")
      }
    })

這種情況下實測消費數(shù)據(jù)入庫速度約2600條每秒。（每條kafka消息約1KiB）

分批寫入HBase

主要變化：創(chuàng)建一個List[Put]，在foreach前創(chuàng)建一個計數(shù)器，不再每條數(shù)據(jù)提交寫一次，而是計數(shù)器每10000時寫一次。
具體多少條提交寫一次，根據(jù)業(yè)務(wù)情況改變。
主要是開頭和結(jié)尾提交時變化，偽代碼如下：

var listPut = new ArrayList[Put]()
var count = 0 
kafkaDStream.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        println("kafkaRDD get some data.")
        rdd.foreachPartition(partitionRecords => {
          // 獲取HBase連接
          val hbaseConnection: Connection = getHBaseConn()
          partitionRecords.foreach(line => {
            // 連接HBase表
            val tableName: TableName = TableName.valueOf(ConfigLoader.getString("hbase.table.name"))
            val table: Table = hbaseConnection.getTable(tableName)
            // 將kafka的每一條消息解析為JSON格式數(shù)據(jù)
            val jsonObj: Option[Any] = JSON.parseFull(line.value())
            val uuid: String = UUID.randomUUID().toString
            // println(line.value())
            val data: Map[String, Any] = jsonObj.get.asInstanceOf[Map[String, Any]]
            val a: String = data("a").asInstanceOf[String]
            val b: String = data("b").asInstanceOf[String]
            val c: String = data("c").asInstanceOf[String]

            val put = new Put(Bytes.toBytes(uuid))
            val tableColumnFamily = ConfigLoader.getString("hbase.table.column.family")
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("a"), Bytes.toBytes(a))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("b"), Bytes.toBytes(b))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("c"), Bytes.toBytes(c))

            // 每次計數(shù)+1
          count +=1
          listPut.add(put)
          if(count % 10000 == 0){
            Try(table.put(listPut)).getOrElse(table.close())
            listPut.clear()
            count = 0
          }
          })
          Try(table.put(listPut)).getOrElse(table.close())
          hbaseConnection.close()
        })
      } else {
        println("kafkaRDD is Empty!!")
      }
    })

這個可以提高入庫速度，具體沒有測。

使用原生批量寫入方法saveAsHadoopDataset

val input = kafkaDStream.flatMap(line=>{
      Some(line.value.toString)
    })
    input.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        println("kafkaRDD get some data.")

        if(args(0).toInt == 0){
          val spark1 = SparkSession.builder().getOrCreate()
          val df = spark1.read.json(rdd)
          df.createOrReplaceTempView("temp")
          val ans = spark1.sql("select a,b,c from temp").rdd.map(x => {
            (UUID.randomUUID.toString, x.getString(0), x.getString(1),x.getString(2),)
          })
          ans.map(line=>{
            val put = new Put(Bytes.toBytes(line._1))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("a"), Bytes.toBytes(line._2))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("b"), Bytes.toBytes(line._3))
            put.addColumn(Bytes.toBytes(tableColumnFamily), Bytes.toBytes("c"), Bytes.toBytes(line._4))
            (new ImmutableBytesWritable, put)
          }).saveAsHadoopDataset(jobConf)
        } else {
          println("接受到:"+rdd.count())
        }
      } else {
        println("kafkaRDD is Empty!!")
      }
    })

這相當(dāng)于調(diào)用RDD.saveAsHadoopDataset(jobConf)，這就不需要自己去處理每多少條數(shù)據(jù)提交寫一次了，后臺使用直接寫Hadoop File的方式。
實測速度提升到13500條數(shù)據(jù)每秒，速度是原來的5.1倍。

點：spark streaming、Hbase、寫入速度調(diào)優(yōu)、流計算
線：Spark
面：內(nèi)存計算