在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>
注意:连接池中的连接应该是,应需求而延迟创建,并且,如果一段时间没用,就超时了(也就是关闭该连接)。