我正在使用Java中的MapReduce和Bulkload将数据加载到Hbase中。基本上，我创建了一个Mapper，并使用HFileOutputFormat2.configureIncrementalLoad（完整代码在问题结尾），用于reduce。我使用一个仅从文件中读取一些字节并创建put的mapper。然后使用LoadIncrementalHFiles.doBulkLoad将数据写入Hbase。这一切都很顺利。但是当执行此操作时，它会覆盖Hbase中旧的值。因此，我正在寻找一种可以追加数据的方法，就像API中的append函数一样。感谢您阅读，希望你们中的一些人能够帮助我 :)
public int run(String[] args) throws Exception {
int result=0;
String outputPath = args[1];
Configuration configuration = getConf();
configuration.set("data.seperator", DATA_SEPERATOR);
configuration.set("hbase.table.name",TABLE_NAME);
configuration.set("COLUMN_FAMILY_1",COLUMN_FAMILY_1);
configuration.set("COLUMN_FAMILY_2",COLUMN_FAMILY_2);
Job job = Job.getInstance(configuration);
job.setJarByClass(HBaseBulkLoadDriver.class);
job.setJobName("Bulk Loading HBase Table::"+TABLE_NAME);
job.setInputFormatClass(TextInputFormat.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapperClass(HBaseBulkLoadMapper.class);
FileInputFormat.addInputPaths(job, args[0]);
FileSystem.getLocal(getConf()).delete(new Path(outputPath), true);
HFileOutputFormat2.setOutputPath(job,new Path((outputPath)));
job.setMapOutputValueClass(Put.class);
Connection c = ConnectionFactory.createConnection(configuration);
Table t = c.getTable(TableName.valueOf(TABLE_NAME));
RegionLocator rl = c.getRegionLocator(TableName.valueOf(TABLE_NAME));
HFileOutputFormat2.configureIncrementalLoad(job,t,rl);
System.out.println("start");
job.waitForCompletion(true);
if (job.isSuccessful()) {
HBaseBulkLoad.doBulkLoad(outputPath, TABLE_NAME);
} else {
result = -1;
}
return result;
}
public static void doBulkLoad(String pathToHFile, String tableName) {
try {
Configuration configuration = new Configuration();
configuration.set("mapreduce.child.java.opts", "-Xmx1g");
HBaseConfiguration.addHbaseResources(configuration);
LoadIncrementalHFiles loadFfiles = new LoadIncrementalHFiles(configuration);
//HTable hTable = new HTable(configuration, tableName);
//loadFfiles.doBulkLoad(new Path(pathToHFile), hTable);
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf(tableName));
Admin admin = connection.getAdmin();
RegionLocator regionLocator = connection.getRegionLocator(TableName.valueOf(tableName));
//path, admin, table, region locator
loadFfiles.doBulkLoad(new Path(pathToHFile),admin,table,regionLocator);
System.out.println("Bulk Load Completed..");
} catch(Exception exception) {
exception.printStackTrace();
}
根据评论中的要求，我在此处添加了表描述的输出，因为该表是通过Python Happybase API创建的，我不确定API默认设置了哪些选项标志...
``` {NAME => '0', BLOOMFILTER => 'NONE', VERSIONS => '3', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'false', BLOCKSIZE => '65536', REPLICATION_SCOPE => '0'} {NAME => '1', BLOOMFILTER => 'NONE', VERSIONS => '3', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK_ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'false', BLOCKSIZE => '65536', REPLICATION_SCOPE => '0'} ```