是否有可用的API使用Spark Scala删除特定的HBase单元。我们能够使用Spark-HBase Connector进行读写。任何有关细胞缺失的建议都是非常可取的。
这是一个Cell
使用Spark删除HBase对象的实现(我已经使用进行了演示parallelize
,您可以将其调整为Cells RDD)。
总体思路:逐块删除-遍历每个RDD分区,将该分区拆分为10,000个单元的块,将每个单元转换为HBaseDelete
对象,然后调用table.delete()
以从HBase进行删除。
public void deleteCells(List<Cell> cellsToDelete) {
JavaSparkContext sc = new JavaSparkContext();
sc.parallelize(cellsToDelete)
.foreachPartition(cellsIterator -> {
int chunkSize = 100000; // Will contact HBase only once per 100,000 records
org.apache.hadoop.conf.Configuration config = new org.apache.hadoop.conf.Configuration();
config.set("hbase.zookeeper.quorum", "YOUR-ZOOKEEPER-HOSTNAME");
Table table;
try {
Connection connection = ConnectionFactory.createConnection(config);
table = connection.getTable(TableName.valueOf(config.get("YOUR-HBASE-TABLE")));
}
catch (IOException e)
{
logger.error("Failed to connect to HBase due to inner exception: " + e);
return;
}
// Split the given cells iterator to chunks
Iterators.partition(cellsIterator, chunkSize)
.forEachRemaining(cellsChunk -> {
List<Delete> deletions = Lists.newArrayList(cellsChunk
.stream()
.map(cell -> new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())
.addColumn(cell.getFamily(), cell.getQualifier(), System.currentTimeMillis()))
.iterator());
try {
table.delete(deletions);
} catch (IOException e) {
logger.error("Failed to delete a chunk due to inner exception: " + e);
}
});
});
}
免责声明:这个确切的代码片段未经测试,但是我使用相同的方法通过Spark删除了数十亿个HBase Cell。
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句