最近工作需要使用到Spark操作Hbase,上篇文章已经写了如何使用Spark读写Hbase全量表的数据做处理,但这次有所不同,这次的需求是Scan特定的Hbase的数据然后转换成RDD做后续处理,简单的使用Google查询了一下,发现实现方式还是比较简单的,用的还是Hbase的TableInputFormat相关的API。
基础软件版本如下:
Hadoop2.7.2
Hbase1.2.0
Spark2.1.0
Scala2.11.8
直接上代码如下:
` val startRowkey="row1"
val endRowkey="row1"
//开始rowkey和结束一样代表精确查询某条数据
//组装scan语句
val scan=new Scan(Bytes.toBytes(startRowkey),Bytes.toBytes(endRowkey))
scan.setCacheBlocks(false)
scan.addFamily(Bytes.toBytes("ks"));
scan.addColumn(Bytes.toBytes("ks"), Bytes.toBytes("data"))
//将scan类转化成string类型
val scan_str= TableMapReduceUtil.convertScanToString(scan)
conf.set(TableInputFormat.SCAN,scan_str)
//使用new hadoop api,读取数据,并转成rdd
val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
//打印扫描的数据总量
println("count:"+rdd.count)
上面的少量代码,已经完整实现了使用spark查询hbase特定的数据,然后统计出数量最后输出,当然上面只是一个简单的例子,重要的是能把hbase数据转换成RDD,只要转成RDD我们后面就能进行非常多的过滤操作。
注意上面的hbase版本比较新,如果是比较旧的hbase,如果自定义下面的方法将scan对象给转成字符串,代码如下:
import java.io.{DataOutputStream, ByteArrayOutputStream}
import java.lang.String
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Base64
def convertScanToString(scan: Scan): String = {
val out: ByteArrayOutputStream = new ByteArrayOutputStream
val dos: DataOutputStream = new DataOutputStream(out)
scan.write(dos)
Base64.encodeBytes(out.toByteArray)
}
最后,还有一点,上面的代码是直接自己new了一个scan对象进行组装,当然我们还可以不自己new对象,全部使用TableInputFormat下面的相关的常量,并赋值,最后执行的时候TableInputFormat会自动帮我们组装scan对象这一点通过看TableInputFormat的源码就能明白:
private static final Log LOG = LogFactory.getLog(TableInputFormat.class);
public static final String INPUT_TABLE = "hbase.mapreduce.inputtable";
private static final String SPLIT_TABLE = "hbase.mapreduce.splittable";
public static final String SCAN = "hbase.mapreduce.scan";
public static final String SCAN_ROW_START = "hbase.mapreduce.scan.row.start";
public static final String SCAN_ROW_STOP = "hbase.mapreduce.scan.row.stop";
public static final String SCAN_COLUMN_FAMILY = "hbase.mapreduce.scan.column.family";
public static final String SCAN_COLUMNS = "hbase.mapreduce.scan.columns";
public static final String SCAN_TIMESTAMP = "hbase.mapreduce.scan.timestamp";
public static final String SCAN_TIMERANGE_START = "hbase.mapreduce.scan.timerange.start";
public static final String SCAN_TIMERANGE_END = "hbase.mapreduce.scan.timerange.end";
public static final String SCAN_MAXVERSIONS = "hbase.mapreduce.scan.maxversions";
public static final String SCAN_CACHEBLOCKS = "hbase.mapreduce.scan.cacheblocks";
public static final String SCAN_CACHEDROWS = "hbase.mapreduce.scan.cachedrows";
public static final String SCAN_BATCHSIZE = "hbase.mapreduce.scan.batchsize";
public static final String SHUFFLE_MAPS = "hbase.mapreduce.inputtable.shufflemaps";
上面代码中的常量,都可以conf.set的时候进行赋值,最后任务运行的时候会自动转换成scan,有兴趣的朋友可以自己尝试。
有什么问题可以扫码关注微信公众号:我是攻城师(woshigcs),在后台留言咨询。 技术债不能欠,健康债更不能欠, 求道之路,与君同行。
分享到:
相关推荐
使用spark读取hbase中的数据,并插入到mysql中
spark使用java读取hbase数据做分布式计算.pdf
基于spark和hbase的电影数据可视化及推荐系统源码(课程设计).zip基于spark和hbase的电影数据可视化及推荐系统源码(课程设计).zip基于spark和hbase的电影数据可视化及推荐系统源码(课程设计).zip基于spark和hbase的...
* Spark:使用spark stream功能,实时分析消息系统中的数据,完成计算分析工作。 * Hbase:做为后端存储,存储spark计算结构,供其他系统进行调用 ## 环境部署 ### 软件版本 * hadoop 版本 : Hadoop相关...
大数据 hadoop spark hbase ambari全套视频教程(购买的付费视频)
hbase-rdd:Spark RDD从HBase读取,写入和删除
通过java api 接口 利用thrift读取hbase表中的数据 解压后包含两个文件 一个是实验步骤文档 一个是java 代码
在这里,我们提供了Scala中的一个新示例,该示例涉及通过Spark将hbase中保存的数据传输到String ,以及python转换器的新示例。 scala 的示例将保存在hbase中的数据传输到RDD[String] ,该数据包含columnFamily,...
Hbase有着先天的优势和先天的劣势,而劣势就是其较差的数据定位能力,也就是数据查询能力。因为面向列的特点,Hbase只能单单地以rowkey为主键作查询,而无法对表进行多维查询和join操作,并且查询通常都是全表扫描,耗费...
最近看了hbase的源码根据源码写了一些scala调动hbase表的API,话不多说直接上代码!Hadoop的版本是2.7.3,scala版本是2.1.1,hbase的版本是1.1.2 如果版本不同可以修改pom的依赖项,但要注意版本冲突。 并且在scala...
java操作Hbase之从Hbase中读取数据写入hdfs中源码,附带全部所需jar包,欢迎下载学习。
:warning: Cloudera Labs中的Spark On HBase项目在2015年被合并到上游HBase项目中,并且自CDH 5.7起CDH附带了Backport SparkOnHBase 概述 这是一个用于Spark的HBase的简单可重用库 功能性 当前功能支持以下功能 批量...
Spark以及hbase学习资料、pdf格式
本文当是一个基于HBase的海量数据的实实时查询系统的原理分析。详细的介绍了大数据查询的原理。
hbase分页查询实现.pdf
spark本地模式访问hbase表进行map reduce操作
用于将 Spark 与 Apache HBase 数据结合使用的集成实用程序。 支持 基于 HBase 读取的扫描 基于 HBase 写入的 batchPut 基于 HBase 读取的分析 HFile 基于 HBase 写入的批量加载 要求 这个库需要 Spark 1.2+
项目以宜信贷风控系统实时数据采集系统为背景,主要描述了技术架构、核心技术难点及代码实现全过程,涉及技术包括但不限于:Kafka\zookeeper,Spark,SparkStreaming,HBase,实时访问技术,爬虫技术等
1、有多个工具可选择使用