如何使用 saveAsNewAPIHadoopDataset 批量插入 hbase

Question

刚学了spark，我用hbase的时候找到了api:saveAsNewAPIHadoopDataset，代码如下，据了解，这段代码可以一次插入一行，怎么改成批量放？我是菜鸟..请帮忙...tks

import org.apache.hadoop.hbase.client.Put  
import org.apache.hadoop.hbase.io.ImmutableBytesWritable  
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat  
import org.apache.hadoop.hbase.client.Result  
import org.apache.hadoop.hbase.util.Bytes  
import org.apache.hadoop.mapreduce.Job  
import org.apache.spark.{SparkContext, SparkConf}  

 /** 
 *
 */  
object HbaseTest2 {  

  def main(args: Array[String]): Unit = {  
    val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")  
    val sc = new SparkContext(sparkConf)  

    val tablename = "account"  

    sc.hadoopConfiguration.set("hbase.zookeeper.quorum","slave1,slave2,slave3")  
    sc.hadoopConfiguration.set("hbase.zookeeper.property.clientPort", "2181")  
    sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE, tablename)  

    val job = Job.getInstance(sc.hadoopConfiguration)  
    job.setOutputKeyClass(classOf[ImmutableBytesWritable])  
    job.setOutputValueClass(classOf[Result])  
    job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])  

    val indataRDD = sc.makeRDD(Array("1,jack,15","2,Lily,16","3,mike,16"))  

    val rdd = indataRDD.map(_.split(',')).map{arr=>{  
      val put = new Put(Bytes.toBytes(arr(0)))  
      put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("name"),Bytes.toBytes(arr(1)))  
      put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("age"),Bytes.toBytes(arr(2).toInt))  
      (new ImmutableBytesWritable, put)  
    }}  

    rdd.saveAsNewAPIHadoopDataset(job.getConfiguration())  

    sc.stop()  
  }  
}  

Answer 1

实际上，您不必担心这一点 - 在后台，put(Put) 和 put(List<Put>) 是相同的。它们都缓冲消息并批量刷新它们。应该没有明显的性能差异。

恐怕另一个答案会被误导。

Answer 2

saveAsNewAPIHadoopDataset 执行单放。

要执行批量放入 hbase 表，您可以使用 hbase-spark 连接器。 连接器在 mapPartition() 中执行 bulkPutFunc2，因此非常高效。 您的源代码将更改如下 -

object HBaseTest {
    def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")
        val sc = new SparkContext(sparkConf)

    val tablename = "account"

    val hbaseConf = HBaseConfiguration.create()
    hbaseConf.set("hbase.zookeeper.quorum", "slave1,slave2,slave3")
    hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")
    hbaseConf.set("zookeeper.znode.parent", "/hbase")

    val hbaseContext = new HBaseContext(sc, hbaseConf)

    val indataRDD = sc.makeRDD(Array("1,jack,15", "2,Lily,16", "3,mike,16"))

    hbaseContext.bulkPut(indataRDD, TableName.valueOf(tablename), bulkPutFunc2)

    sc.stop()
}
def bulkPutFunc2(arrayRec : String): Put = {
    val rec = arrayRec.split(",")
    val put = new Put(Bytes.toBytes(rec(0).toInt))
    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes(rec(1)))
    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("age"), Bytes.toBytes(rec(2).toInt))

    put
    }

}

pom.xml 将有以下条目 -

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-spark</artifactId>
    <version>1.2.0-cdh5.12.1</version>
<dependency>