火花过滤器 hbase 以获取样本答案

【问题标题】：spark filter hbase to get samples火花过滤器 hbase 以获取样本
【发布时间】：2016-09-05 14:21:05
【问题描述】：

我有一个大型的 hbase 数据集，我想检索一些具有特定条件的数据以获取用于调试的样本集。

这里是 Spark RDD 和过滤器：

val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.quorum", "172.16.1.10,172.16.1.11,172.16.1.12")
conf.setInt("timeout", 120000)
conf.set(TableInputFormat.INPUT_TABLE, "dateset")
val hbaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
val filteredRDD = rdd.filter{
   case tuple => {
      val result = tuple._2
      val hostId = new String(result.getValue("user", "id"))
      hostId == "12345" // <-- only retrieve the row when user:id is 12345
   }
}

现在我得到了filteredRDD 的rdd，我只想将它以相同的格式保存在另一个表中

val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.quorum", "172.16.1.10,172.16.1.11,172.16.1.12")
conf.setInt("timeout", 120000)
conf.set(TableOutputFormat.OUTPUT_TABLE, "data_sample")
// here I don't know which api to use

谁能给我一个线索？谢谢。

【问题讨论】：

我已经回答了如何在下面的 Java 中做同样的事情，因为我不是 Scala 专家。希望这也适用于 Scala。
您确定您的代码适用于filteredRDD 吗？它不适合我。它说getValue 方法需要Array[Byte] 而不是字符串。

标签： scala apache-spark hbase

【解决方案1】：

您可以在对 rdd 上使用 saveAsNewAPIHadoopDataset 函数，如下所示：

JavaPairRDD<ImmutableBytesWritable, Put> pairs = filterRDD.mapToPair(new PairFunction<Object, ImmutableBytesWritable, Put>() {
    @Override
    public Tuple2<ImmutableBytesWritable, Put> execute(Object message) throws Exception {

        //prepare the HBase put here
        return new Tuple2<ImmutableBytesWritable, Put>(new ImmutableBytesWritable(), put);
    }
});

Job newAPIJobConfiguration = Job.getInstance(hbaseConf);
newAPIJobConfiguration.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName);
newAPIJobConfiguration.setOutputFormatClass(org.apache.hadoop.hbase.mapreduce.TableOutputFormat.class);

pairs.saveAsNewAPIHadoopDataset(newAPIJobConfiguration.getConfiguration());

【讨论】：