HBase：基本Spark

基本Spark

本節(jié)討論最低和最簡(jiǎn)單級(jí)別的Spark HBase集成。所有其他交互點(diǎn)都建立在此處描述的概念之上。

所有Spark和HBase集成的根源都是HBaseContext。HBaseContext接受HBase配置并將它們推送到Spark執(zhí)行程序。這允許我們?cè)陟o態(tài)位置為每個(gè)Spark Executor建立一個(gè)HBase連接。

作為參考，Spark Executors可以與Region Servers位于相同的節(jié)點(diǎn)上，也可以位于不同的節(jié)點(diǎn)上，不依賴于co-location。將每個(gè)Spark Executor都視為多線程客戶端應(yīng)用程序。這允許在執(zhí)行程序上運(yùn)行的任何Spark任務(wù)訪問(wèn)共享的Connection對(duì)象。

HBaseContext用法示例

本示例演示如何使用HBaseContext在Scala的RDD上執(zhí)行foreachPartition：

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

...

val hbaseContext = new HBaseContext(sc, config)

rdd.hbaseForeachPartition(hbaseContext, (it, conn) => {
 val bufferedMutator = conn.getBufferedMutator(TableName.valueOf("t1"))
 it.foreach((putRecord) => {
. val put = new Put(putRecord._1)
. putRecord._2.foreach((putValue) => put.addColumn(putValue._1, putValue._2, putValue._3))
. bufferedMutator.mutate(put)
 })
 bufferedMutator.flush()
 bufferedMutator.close()
})

這是在Java中實(shí)現(xiàn)的相同示例：

JavaSparkContext jsc = new JavaSparkContext(sparkConf);

try {
  List<byte[]> list = new ArrayList<>();
  list.add(Bytes.toBytes("1"));
  ...
  list.add(Bytes.toBytes("5"));

  JavaRDD<byte[]> rdd = jsc.parallelize(list);
  Configuration conf = HBaseConfiguration.create();

  JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);

  hbaseContext.foreachPartition(rdd,
      new VoidFunction<Tuple2<Iterator<byte[]>, Connection>>() {
   public void call(Tuple2<Iterator<byte[]>, Connection> t)
        throws Exception {
    Table table = t._2().getTable(TableName.valueOf(tableName));
    BufferedMutator mutator = t._2().getBufferedMutator(TableName.valueOf(tableName));
    while (t._1().hasNext()) {
      byte[] b = t._1().next();
      Result r = table.get(new Get(b));
      if (r.getExists()) {
       mutator.mutate(new Put(b));
      }
    }

    mutator.flush();
    mutator.close();
    table.close();
   }
  });
} finally {
  jsc.stop();
}

Scala和Java都支持Spark和HBase之間的所有功能，但SparkSQL除外，它支持Spark支持的任何語(yǔ)言。對(duì)于本文檔的其余部分，我們現(xiàn)在將重點(diǎn)介紹Scala示例。

上面的示例說(shuō)明了如何使用連接執(zhí)行foreachPartition。一些其他的Spark基礎(chǔ)功能是支持開(kāi)箱即用：

bulkPut

用于向HBase大規(guī)模并行發(fā)送put

bulkDelete

用于向HBase大規(guī)模并行發(fā)送delete

bulkGet

用于向HBase大規(guī)模并行發(fā)送get，以創(chuàng)建新的RDD

mapPartition

使用Connection對(duì)象執(zhí)行Spark Map功能以允許完全訪問(wèn)HBase

hBaseRDD

簡(jiǎn)化分布式掃描以創(chuàng)建RDD

有關(guān)所有這些功能的示例，請(qǐng)參閱HBase-Spark模塊。