spark运算结果写入hbase及优化

在Spark中利用map-reduce或者spark sql分析了数据之后，我们需要将结果写入外部文件系统。

本文，以向Hbase中写数据，为例，说一下，Spark怎么向Hbase中写数据。

首先，需要说一下，下面的这个方法。

foreach (func)

最通用的输出操作，把func作用于从map-reduce生成的每一个RDD(spark  sql生成的DataFrame可转成RDD)。

注意：这个函数是在运行spark程序的driver进程中执行的。

下面跟着思路，看一下，怎么优雅的向Hbase中写入数据

向外部数据库写数据，通常会建立连接，使用连接发送数据(也就是保存数据)。

<pre name="code" class="java"><span style="font-size:18px;color:#003300;">DataFrame dataFrame = Contexts.hiveContext.sql("select * from tableName");
dataFrame.javaRDD().foreach(new VoidFunction<Row> () {
    public void call(Row row) {
        HConnection conn = ...
        HTableInterface htable = conn.getTable("");
        //save to hbase
    }
});</span>

很遗憾！这种写法是有极大风险的，这会导致，对于每条数据，都创建一个connection(创建connection是消耗资源的)。

事实上，由于数据是分区的，基于这个特性，还可以有更高效的方式

下面的方法会好一些：

<span style="font-size:18px;color:#003300;">DataFrame dataFrame = Contexts.hiveContext.sql("select * from tableName");
dataFrame.javaRDD().foreachPartition(new VoidFunction<Iterator<Row>> () {
    public void call(Iterator<Row> rows) {
        HConnection conn = ...
        HTableInterface htable = conn.getTable("");
        while(rows.hasNext()){
            //save to hbase
        }
    }
});</span>

上面的方法，使用 rdd.foreachPartition创建一个connection对象，一个RDD分区中的所有数据，都使用这一个connection。

在多个RDD之间，connection对象是可以重用的，所以可以创建一个连接池。如下：

<span style="font-size:18px;color:#003300;">dataFrame.javaRDD().foreachPartition(new VoidFunction<Iterator<Row>> () {
    public void call(Iterator<Row> rows) {
        HTableInterface htable = TablePool.getHTable("");
        while(rows.hasNext()){
            //save to hbase
        }
    }
});</span>

注意：连接池中的连接应该是，应需求而延迟创建，并且，如果一段时间没用，就超时了(也就是关闭该连接)。