Spark批量删除hbase上的记录

Question

我找到了很多关于从 hbase 在 spark 中加载数据的示例，对我有用的是

Configuration config = HBaseConfiguration.create();
    config.set(TableInputFormat.INPUT_TABLE, props.getProperty(ConfigConstants.HBASE_SRC_TABLE_NAME));
    config.set(TableInputFormat.SCAN_MAXVERSIONS, props.getProperty(ConfigConstants.HBASE_SRC_TABLE_VERSIONS));
    config.set(TableInputFormat.SCAN_COLUMN_FAMILY, HbaseConstants.MAPPING_FAMILY);
    config.set(TableInputFormat.SCAN_TIMERANGE_START, "0");
    config.set(TableInputFormat.SCAN_TIMERANGE_END, startTimestamp + "000");

    RDD<Tuple2<ImmutableBytesWritable, Result>> tupleRDD = context.newAPIHadoopRDD(config, TableInputFormat.class,
            ImmutableBytesWritable.class, Result.class);

但是我真的需要一种方法来删除处理后加载到 spark 中的记录。

尝试将 tupleRDD 映射到 JavaPairRDD<ImmutableBytesWritable, Delete>，然后使用

JobConf jobConf = new JobConf(config);
    jobConf.setOutputFormat(org.apache.hadoop.hbase.mapred.TableOutputFormat.class);
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, props.getProperty(ConfigConstants.HBASE_TARGET_TABLE_NAME));
    outputPairsRDD.saveAsHadoopDataset(jobConf);

但这给了我一个例外，如下所示

"main" org.apache.hadoop.mapred.InvalidJobConfException: Output directory not set.

有没有办法从 spark 中删除？

Answer 1

原来没有简单的方法可以做到这一点，我最终的解决方案是通过hbase delete函数而不是hbase的spark-version delete来抓取数据并删除它们。

Spark批量删除hbase上的记录

Spark batch delete records on hbase

hbase

apache-spark