正文
包含hbasespark的词条
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
如何提高spark批量读取HBase数据的性能
Configuration conf = HBaseConfiguration.create();
String tableName = "testTable";
Scan scan = new Scan();
scan.setCaching(10000);
scan.setCacheBlocks(false);
conf.set(TableInputFormat.INPUT_TABLE, tableName);
ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
String ScanToString = Base64.encodeBytes(proto.toByteArray());
conf.set(TableInputFormat.SCAN, ScanToString);
JavaPairRDDImmutableBytesWritable, Result myRDD = sc
.newAPIHadoopRDD(conf, TableInputFormat.class,
ImmutableBytesWritable.class, Result.class);
在Spark使用如上Hadoop提供的标准接口读取HBase表数据（全表读），读取5亿左右数据，要20M+，而同样的数据保存在Hive中，读取却只需要1M以内，性能差别非常大。
转载，仅供参考。
相关问答
Q1: Spark 读取 Hbase 数据
下面这种方式是全表扫描，Spark如果通过RS来访问Hbase数据进行数据分析，对RS会产生很大的压力。不太建议使用下面的方式
在本地测试时返现运行的很慢，后来看到以下日志
由于Hbase表中只有两个region，所以只启动两个Task，此时并行度为二！
那么也就是说Spark读取Hbase的并行度取决于这个表有多少个region。然后根据region的startkey和endkey来获取数据
Q2: 如何使用scala+spark读写hbase
如何使用scala+spark读写Hbase
软件版本如下：
scala2.11.8
spark2.1.0
hbase1.2.0
公司有一些实时数据处理的项目，存储用的是hbase，提供实时的检索，当然hbase里面存储的数据模型都是简单的，复杂的多维检索的结果是在es里面存储的，公司也正在引入Kylin作为OLAP的数据分析引擎，这块后续有空在研究下。
接着上面说的，hbase存储着一些实时的数据，前两周新需求需要对hbase里面指定表的数据做一次全量的update以满足业务的发展，平时操作hbase都是单条的curd，或者插入一个批量的list，用的都是hbase的java api比较简单，但这次涉及全量update，所以如果再用原来那种单线程的操作api，势必速度回慢上许多。
关于批量操作Hbase，一般我们都会用MapReduce来操作，这样可以大大加快处理效率，原来也写过MR操作Hbase，过程比较繁琐，最近一直在用scala做spark的相关开发，所以就直接使用scala+spark来搞定这件事了，当然底层用的还是Hbase的TableOutputFormat和TableOutputFormat这个和MR是一样的，在spark里面把从hbase里面读取的数据集转成rdd了，然后做一些简单的过滤，转化，最终在把结果写入到hbase里面。
整个流程如下：
（1）全量读取hbase表的数据
（2）做一系列的ETL
（3）把全量数据再写回hbase
核心代码如下：
//获取conf
val conf=HBaseConfiguration.create() //设置读取的表
conf.set(TableInputFormat.INPUT_TABLE,tableName) //设置写入的表
conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)//创建sparkConf
val sparkConf=new SparkConf() //设置spark的任务名
sparkConf.setAppName("read and write for hbase ") //创建spark上下文
val sc=new SparkContext(sparkConf)
//为job指定输出格式和输出表名
val newAPIJobConfiguration1 = Job.getInstance(conf)
newAPIJobConfiguration1.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName)
newAPIJobConfiguration1.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
//全量读取hbase表
val rdd=sc.newAPIHadoopRDD(conf,classOf[TableInputFormat]
,classOf[ImmutableBytesWritable]
,classOf[Result]
)
//过滤空数据，然后对每一个记录做更新，并转换成写入的格式
val final_rdd= rdd.filter(checkNotEmptyKs).map(forDatas)
//转换后的结果，再次做过滤
val save_rdd=final_rdd.filter(checkNull)
//最终在写回hbase表
save_rdd.saveAsNewAPIHadoopDataset(newAPIJobConfiguration1.getConfiguration)
sc.stop()
从上面的代码可以看出来，使用spark+scala操作hbase是非常简单的。下面我们看一下，中间用到的几个自定义函数：
第一个：checkNotEmptyKs
作用：过滤掉空列簇的数据
def checkNotEmptyKs(f:((ImmutableBytesWritable,Result))):Boolean={ val r=f._2 val rowkey=Bytes.toString(r.getRow) val map:scala.collection.mutable.Map[Array[Byte],Array[Byte]]= r.getFamilyMap(Bytes.toBytes("ks")).asScala if(map.isEmpty) false else true
}
第二个：forDatas
作用：读取每一条数据，做update后，在转化成写入操作
def forDatas(f: (ImmutableBytesWritable,Result)): (ImmutableBytesWritable,Put)={ val r=f._2 //获取Result
val put:Put=new Put(r.getRow) //声明put
val ks=Bytes.toBytes("ks") //读取指定列簇
val map:scala.collection.mutable.Map[Array[Byte],Array[Byte]]= r.getFamilyMap(ks).asScala
map.foreach(kv={//遍历每一个rowkey下面的指定列簇的每一列的数据做转化
val kid= Bytes.toString(kv._1)//知识点id
var value=Bytes.toString(kv._2)//知识点的value值
value="修改后的value"
put.addColumn(ks,kv._1,Bytes.toBytes(value)) //放入put对象
}
) if(put.isEmpty) null else (new ImmutableBytesWritable(),put)
}
第三个：checkNull 作用：过滤最终结果里面的null数据
def checkNull(f:((ImmutableBytesWritable,Put))):Boolean={ if(f==null) false else true
}
上面就是整个处理的逻辑了，需要注意的是对hbase里面的无效数据作过滤，跳过无效数据即可，逻辑是比较简单的，代码量也比较少。
除了上面的方式，还有一些开源的框架，也封装了相关的处理逻辑，使得spark操作hbase变得更简洁，有兴趣的朋友可以了解下，github链接如下：
Q3: 如何使用Spark/Scala读取Hbase的数据
必须使用高亮参数启Spark-shell,否则遍历RDD现Exception
java.io.NotSerializableException: org.apache.hadoop.hbase.io.ImmutableBytesWritable
spark-shell--conf spark.serializer=org.apache.spark.serializer.KryoSerializer
代码,经MaprDB实测通
import org.apache.spark._
import org.apache.spark.rdd.NewHadoopRDD
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HColumnDescriptor
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.HTable;
val tableName = "/app/SubscriptionBillingPlatform/TRANSAC_ID"
val conf = HBaseConfiguration.create()
conf.set(TableInputFormat.INPUT_TABLE, tableName)
//create rdd
val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
hBaseRDD.take(2).map(row=row._2.rawCells).
map(_.map( kv = (new String(kv.getQualifier()) - new
String(kv.getValue()) ) ).toMap ). foreach( map = { map.foreach{
entry = print(entry._1 +":" + entry._2 + ", ") } ;
print("\n-----------\n") } )
//get the row count
val count = hBaseRDD.count()
print("HBase RDD count:"+count)
关于hbasespark和的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。
