Spark读取HbaseRDD的过程及源码解析
需积分: 1 170 浏览量
2023-04-07
17:16:33
上传
评论
收藏 136KB PDF 举报
浪尖 qq 技术交流群 459898801 224209501
浪尖 qq 技术交流群 459898801 224209501
Spark 读取 HbaseRDD 的过程
1,构建 hbaseRDD
val hbaseRDD = sc.newAPIHadoopRDD(confHbase, classOf[TableInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
2,构建 NewHadoopRDD
new NewHadoopRDD(this, fClass, kClass, vClass, jconf)
3,getPartitions 方法
Table 的初始化
val inputFormat = inputFormatClass.newInstance
inputFormat match {
case configurable: Configurable =>
configurable.setConf(_conf)
case _ =>
}
TableInputFormat 的 setConf 方法，给 InputFormat 设置表
this.conf = configuration;
String tableName = conf.get(
INPUT_TABLE
);
try {
setHTable(new HTable(new Configuration(conf), tableName));
} catch (Exception e) {
LOG.error(StringUtils.
stringifyException
(e));
}
初始化 scan，会先会获取用户设置的 scan 方法，然后再跟进多种情况进行构建。
Scan scan = null;
if (conf.get(
SCAN
) != null) {
try {
scan = TableMapReduceUtil.
convertStringToScan
(conf.get(
SCAN
));
} catch (IOException e) {
LOG.error("An error occurred.", e);
}
} else {
try {
scan = new Scan();
if (conf.get(
SCAN_ROW_START
) != null) {
scan.setStartRow(Bytes.
toBytes
(conf.get(
SCAN_ROW_START
)));
}
if (conf.get(
SCAN_ROW_STOP
) != null) {
scan.setStopRow(Bytes.
toBytes
(conf.get(
SCAN_ROW_STOP
)));
}
资源评论
