spark运算结果写入hbase及优化
2023-09-27 14:26:54 时间
在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>
注意:连接池中的连接应该是,应需求而延迟创建,并且,如果一段时间没用,就超时了(也就是关闭该连接)。
相关文章
- HIVE HBASE 整合
- kylin: build cube Hbase: Region Server 意外退出
- 欢迎加入阿里云 HBase+Spark技术交流群
- HBase-TDG ClientAPI The Basics
- HBase Phoenix助力海量数据实时分析
- HBase参数配置及说明
- 《HBase权威指南》一第3章 客户端API:基础知识
- HBase(二)CentOS7.5搭建HBase1.2.6HA集群
- HBase简介
- HBase与MongDB等NoSQL数据库对照
- 《HBase企业应用开发实战》—— 2.1 先决条件
- 《HBase企业应用开发实战》—— 3.4 数据模型的特殊属性
- HBase性能优化方法总结(转)
- 用Spark查询HBase中的表数据
- Hbase存储思想
- 【大数据监控】Grafana、Spark、HDFS、YARN、Hbase指标性能监控安装部署详细文档
- 统计hbase中某张表中数据条数