Spark 下操作 HBase
HBase经过七年发展,终于在今年2月底,发布了 1.0.0 版本。这个版本提供了一些让人激动的功能,并且,在不牺牲稳定性的前提下,引入了新的API。虽然 1.0.0 兼容旧版本的 API,不过还是应该尽早地来熟悉下新版API。并且了解下如何与当下正红的 Spark 结合,进行数据的写入与读取。鉴于国内外有关 HBase 1.0.0 新 API 的资料甚少,故作此文。
本文将分两部分介绍,第一部分讲解使用 HBase 新版 API 进行 CRUD 基本操作;第二部分讲解如何将 Spark 内的 RDDs 写入 HBase 的表中,反之,HBase 中的表又是如何以 RDDs 形式加载进 Spark 内的。
环境配置
为了避免版本不一致带来不必要的麻烦,API 和 HBase环境都是 1.0.0 版本。HBase 为单机模式,分布式模式的使用方法类似,只需要修改HBaseConfiguration
的配置即可。
开发环境中使用 SBT 加载依赖项
name := "SparkLearn" version := "1.0" scalaVersion := "2.10.4" libraryDependencies += "org.apache.spark" %% "spark-core" % "1.3.0" libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.0.0" libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.0.0" libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.0.0" |
HBase 的 CRUD 操作
新版 API 中加入了 Connection
,HAdmin
成了Admin
,HTable
成了Table
,而Admin
和Table
只能通过Connection
获得。Connection
的创建是个重量级的操作,由于Connection
是线程安全的,所以推荐使用单例,其工厂方法需要一个HBaseConfiguration
。
val conf = HBaseConfiguration.create() conf.set("hbase.zookeeper.property.clientPort", "2181") conf.set("hbase.zookeeper.quorum", "master") //Connection 的创建是个重量级的工作,线程安全,是操作hbase的入口 val conn = ConnectionFactory.createConnection(conf) |
创建表
使用Admin
创建和删除表
val userTable = TableName.valueOf("user") //创建 user 表 val tableDescr = new HTableDescriptor(userTable) tableDescr.addFamily(new HColumnDescriptor("basic".getBytes)) println("Creating table `user`. ") if (admin.tableExists(userTable)) { admin.disableTable(userTable) admin.deleteTable(userTable) } admin.createTable(tableDescr) println("Done!") |
插入、查询、扫描、删除操作
HBase 上的操作都需要先创建一个操作对象Put
,Get
,Delete
等,然后调用Table
上的相对应的方法
try{ //获取 user 表 val table = conn.getTable(userTable) try{ //准备插入一条 key 为 id001 的数据 val p = new Put("id001".getBytes) //为put操作指定 column 和 value (以前的 put.add 方法被弃用了) p.addColumn("basic".getBytes,"name".getBytes, "wuchong".getBytes) //提交 table.put(p) //查询某条数据 val g = new Get("id001".getBytes) val result = table.get(g) val value = Bytes.toString(result.getValue("basic".getBytes,"name".getBytes)) println("GET id001 :"+value) //扫描数据 val s = new Scan() s.addColumn("basic".getBytes,"name".getBytes) val scanner = table.getScanner(s) try{ for(r <- scanner){ println("Found row: "+r) println("Found value: "+Bytes.toString( r.getValue("basic".getBytes,"name".getBytes))) } }finally { //确保scanner关闭 scanner.close() } //删除某条数据,操作方式与 Put 类似 val d = new Delete("id001".getBytes) d.addColumn("basic".getBytes,"name".getBytes) table.delete(d) }finally { if(table != null) table.close() } }finally { conn.close() } |
Spark 操作 HBase
写入 HBase
首先要向 HBase 写入数据,我们需要用到PairRDDFunctions.saveAsHadoopDataset
。因为 HBase 不是一个文件系统,所以saveAsHadoopFile
方法没用。
def saveAsHadoopDataset(conf: JobConf): Unit
Output the RDD to any Hadoop-supported storage system, using a Hadoop JobConf object for that storage system
这个方法需要一个 JobConf 作为参数,类似于一个配置项,主要需要指定输出的格式和输出的表名。
Step 1:我们需要先创建一个 JobConf。
//定义 HBase 的配置 val conf = HBaseConfiguration.create() conf.set("hbase.zookeeper.property.clientPort", "2181") conf.set("hbase.zookeeper.quorum", "master") //指定输出格式和输出表名 val jobConf = new JobConf(conf,this.getClass) jobConf.setOutputFormat(classOf[TableOutputFormat]) jobConf.set(TableOutputFormat.OUTPUT_TABLE,"user") |
Step 2: RDD 到表模式的映射
在 HBase 中的表 schema 一般是这样的:
row cf:col_1 cf:col_2
而在Spark中,我们操作的是RDD元组,比如(1,"lilei",14)
, (2,"hanmei",18)
。我们需要将 RDD[(uid:Int, name:String, age:Int)]
转换成 RDD[(ImmutableBytesWritable, Put)]
。所以,我们定义一个 convert 函数做这个转换工作
def convert(triple: (Int, String, Int)) = { val p = new Put(Bytes.toBytes(triple._1)) p.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("name"),Bytes.toBytes(triple._2)) p.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("age"),Bytes.toBytes(triple._3)) (new ImmutableBytesWritable, p) } |
Step 3: 读取RDD并转换
//read RDD data from somewhere and convert val rawData = List((1,"lilei",14), (2,"hanmei",18), (3,"someone",38)) val localData = sc.parallelize(rawData).map(convert) |
Step 4: 使用saveAsHadoopDataset
方法写入HBase
localData.saveAsHadoopDataset(jobConf) |
读取 HBase
Spark读取HBase,我们主要使用SparkContext
提供的newAPIHadoopRDD
API将表的内容以 RDDs 的形式加载到 Spark 中。
val conf = HBaseConfiguration.create() conf.set("hbase.zookeeper.property.clientPort", "2181") conf.set("hbase.zookeeper.quorum", "master") //设置查询的表名 conf.set(TableInputFormat.INPUT_TABLE, "user") val usersRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result]) val count = usersRDD.count() println("Users RDD Count:" + count) usersRDD.cache() //遍历输出 usersRDD.foreach{ case (_,result) => val key = Bytes.toInt(result.getRow) val name = Bytes.toString(result.getValue("basic".getBytes,"name".getBytes)) val age = Bytes.toInt(result.getValue("basic".getBytes,"age".getBytes)) println("Row key:"+key+" Name:"+name+" Age:"+age) } |
附录
更完整的代码已上传到 Gist 。
- HBaseNewAPI.scala HBase 的 CRUD 操作
- SparkOnHBase.scala Spark 操作 HBase
Spark 下操作 HBase相关推荐
- hbase linux 命令,在linux下操作hbase
在linux下操作hbase作者:bin 这个月比较忙,一直都没有时间整理最近学习的东西,T_T 这里介绍如何使用hbase shell进行操作hbase 具体的安装.配置,可以随意在网上找到,这里不 ...
- c# 访问hbase_大数据技术 windows下C#通过Thrift操作HBase
本篇教程探讨了大数据技术 windows下C#通过Thrift操作HBase,希望阅读本篇文章以后大家有所收获,帮助大家对大数据技术的理解更加深入. < 1.到apache官网下载Thrift源 ...
- HBase实战(6):使用Spark 2.2.1 直接操作HBASE 1.2.0数据库
HBase实战(6):使用Spark 2.2.1 直接操作HBASE 1.2.0数据库 之前对于Hbase系统已实验成功的内容: Hbase分布式集群搭建:点击打开链接 直接使用python API连 ...
- Spark SQL操作之-函数汇总篇-下
Spark SQL操作之-自定义函数篇-下 环境说明 自定义函数分类 用户自定义函数(UDF) 用户自定义聚合函数(UDAF) 环境说明 1. JDK 1.8 2. Spark 2.1 自定义函数分类 ...
- HBase安装配置以及Java操作hbase
2019独角兽企业重金招聘Python工程师标准>>> Apache HBase Apache HBase™是Hadoop数据库,是一个分布式,可扩展的大数据存储. 当您需要对大数据 ...
- Spark RDD/Core 编程 API入门系列之动手实战和调试Spark文件操作、动手实战操作搜狗日志文件、搜狗日志文件深入实战(二)...
1.动手实战和调试Spark文件操作 这里,我以指定executor-memory参数的方式,启动spark-shell. 启动hadoop集群 spark@SparkSingleNode:/usr/ ...
- hive删除hbase数据_Hive进阶:Hive通过外部表操作Hbase数据
概述: HBase: 查询效率比较高,常为实时业务提供服务,但是其查询方式比较单一,只能通过row方式get单条数据,或者通过scan加过滤器的方式扫描数据表获取数据. Hive: hive用来存储结 ...
- PHP通过Thrift操作Hbase
HBase是一个开源的NoSQL产品,它是实现了Google BigTable论文的一个开源产品,和Hadoop和HDFS一起,可用来存储和处理海量column family的数据.官方网址是:htt ...
- HBase 6、用Phoenix Java api操作HBase
开发环境准备:eclipse3.5.jdk1.7.window8.hadoop2.2.0.hbase0.98.0.2.phoenix4.3.0 1.从集群拷贝以下文件:core-site.xml.hb ...
最新文章
- Arcgis10.2安装与解决 [转载自麻辣GIS]
- mysql表结构 转 golang 结构体struct
- 详解KMP算法原理,以及完整java与C++实现
- php动态删除输入框,jQuery实现动态添加和删除input框实例代码
- Hive _分桶及抽样查询
- c++MMMMM:oo
- 什么是5G,我们能从中得到什么?
- 获取referer中的请求参数_Servlet获取AJAX POST请求中参数以form data和request payload形式传输的方法...
- One-hot encoding 独热编码
- Android异步批量下载图片并缓存
- 【概率论】5-9:多项式分布(The Multinomial Distributions)
- C语言入门学习(入门级C语言)
- make_interp_spline(x, y[1:151])(x_smooth) ValueError: x and y are incompatible.
- jQuery常用功能大全
- java简易计算器报告_JAVA实训报告简易计算器.doc
- HEVC官方软件HM源代码简单分析-解码器TAppDecoder
- 如何创建XS Job来完成定时任务
- 通俗理解LDA主题模型(转)
- Pytorch中nn.Module中的self.register_buffer解释
- 微信小程序调用地图设置起点终点导航
热门文章
- Nginx配置统计页面及访问控制(htpasswd和客户端IP)
- shell编程之条件语句(文件测试,test命令,字符串和逻辑测试,if单支语句,if双支语句,if多支语句,case命令,用if写跑步小实验)
- spring boot plugin_spring-boot-starter-parent 与 spring-boot-dependencies
- Java 判断文件夹、文件是否存在、否则创建文件夹
- 马云马化腾,过的哪个冬
- eclipse/myeclipse高亮显示相同变量名 .
- java j2se1.5_用J2SE1.5建立多任务的Java应用程序...
- synchronized原理_synchronized关键字的作用、原理以及锁优化
- pve 不订阅更新_??“吃鸡”体验服已无更新,暗夜危机2.0或将被1款新游代替
- webpack html转成js,WebPack的基础学习