Spark和HBase快照
假设我们可以从HDFS直接获取数据而不是使用HBase API来更快地访问数据，我们正在尝试基于HBase的Table Snapshot构建RDD。
所以，我有一个叫做“dm_test_snap”的快照。 我似乎能够获得大部分配置工作，但是我的RDD为空（尽管Snapshot中有数据）。
我有一段时间找到了一个人用Spark来对HBase快照进行离线分析的例子，但我无法相信我会一个人努力做到这一点。 任何帮助或建议非常感谢。
这是我的代码片段：
object TestSnap {
def main(args: Array[String]) {
val config = ConfigFactory.load()
val hbaseRootDir = config.getString("hbase.rootdir")
val sparkConf = new SparkConf()
.setAppName("testnsnap")
.setMaster(config.getString("spark.app.master"))
.setJars(SparkContext.jarOfObject(this))
.set("spark.executor.memory", "2g")
.set("spark.default.parallelism", "160")
val sc = new SparkContext(sparkConf)
println("Creating hbase configuration")
val conf = HBaseConfiguration.create()
conf.set("hbase.rootdir", hbaseRootDir)
conf.set("hbase.zookeeper.quorum", config.getString("hbase.zookeeper.quorum"))
conf.set("zookeeper.session.timeout", config.getString("zookeeper.session.timeout"))
conf.set("hbase.TableSnapshotInputFormat.snapshot.name", "dm_test_snap")
val scan = new Scan
val job = Job.getInstance(conf)
TableSnapshotInputFormat.setInput(job, "dm_test_snap",
new Path("hdfs://nameservice1/tmp"))
val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableSnapshotInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
hBaseRDD.count()
System.exit(0)
}
}
更新以包含解决方案诀窍是，正如@Holden下面提到的那样，conf并没有通过。 为了解决这个问题，我可以通过将这个调用改为newAPIHadoopRDD来实现：
val hBaseRDD = sc.newAPIHadoopRDD(job.getConfiguration, classOf[TableSnapshotInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
还有第二个问题，也是@维克特的答案突出的，我没有通过扫描。 为了解决这个问题，我添加了这一行和方法：
conf.set(TableInputFormat.SCAN, convertScanToString(scan))
def convertScanToString(scan : Scan) = {
val proto = ProtobufUtil.toScan(scan);
Base64.encodeBytes(proto.toByteArray());
}
这也让我从conf.set命令中拉出这一行：
conf.set("hbase.TableSnapshotInputFormat.snapshot.name", "dm_test_snap")
*注意：这是针对CDH5.0上的HBase 0.96.1.1版本的
最后完整的代码以便于参考：
object TestSnap {
def main(args: Array[String]) {
val config = ConfigFactory.load()
val hbaseRootDir = config.getString("hbase.rootdir")
val sparkConf = new SparkConf()
.setAppName("testnsnap")
.setMaster(config.getString("spark.app.master"))
.setJars(SparkContext.jarOfObject(this))
.set("spark.executor.memory", "2g")
.set("spark.default.parallelism", "160")
val sc = new SparkContext(sparkConf)
println("Creating hbase configuration")
val conf = HBaseConfiguration.create()
conf.set("hbase.rootdir", hbaseRootDir)
conf.set("hbase.zookeeper.quorum", config.getString("hbase.zookeeper.quorum"))
conf.set("zookeeper.session.timeout", config.getString("zookeeper.session.timeout"))
val scan = new Scan
conf.set(TableInputFormat.SCAN, convertScanToString(scan))
val job = Job.getInstance(conf)
TableSnapshotInputFormat.setInput(job, "dm_test_snap",
new Path("hdfs://nameservice1/tmp"))
val hBaseRDD = sc.newAPIHadoopRDD(job.getConfiguration, classOf[TableSnapshotInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
hBaseRDD.count()
System.exit(0)
}
def convertScanToString(scan : Scan) = {
val proto = ProtobufUtil.toScan(scan);
Base64.encodeBytes(proto.toByteArray());
}
}
查看作业信息，它将制作你提供给它的conf对象的副本（ The Job makes a copy of the Configuration so that any necessary internal modifications do not reflect on the incoming parameter.因此，最有可能的信息是你需要在conf对象上设置不会传递给Spark。 你可以改用TableSnapshotInputFormatImpl ，它有一个类似于conf对象的方法。 可能还需要额外的东西，但首先通过这个问题似乎是最可能的原因。
正如在注释中指出的那样，另一种选择是使用job.getConfiguration从作业对象获取更新的配置。
您尚未正确配置您的M / R作业：这是Java中关于如何配置M / R over snapshots的示例：
Job job = new Job(conf);
Scan scan = new Scan();
TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
scan, MyTableMapper.class, MyMapKeyOutput.class,
MyMapOutputValueWritable.class, job, true);
}
你肯定跳过了扫描。 我建议你看看TableMapReduceUtil的initTableSnapshotMapperJob实现，以了解如何在Spark / Scala中配置作业。
这是mapreduce Java中的完整配置
TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, // Name of the snapshot
scan, // Scan instance to control CF and attribute selection
DefaultMapper.class, // mapper class
NullWritable.class, // mapper output key
Text.class, // mapper output value
job,
true,
restoreDir);