org.apache.hadoop.hbase.client.BufferedMutator主要用来对HBase的单个表进行操作。它和Put类的作用差不多，但是主要用来实现批量的异步写操作。
BufferedMutator替换了HTable的setAutoFlush(false)的作用。
可以从Connection的实例中获取BufferedMutator的实例。在使用完成后需要调用close()方法关闭连接。对BufferedMutator进行配置需要通过BufferedMutatorParams完成。
MapReduce Job的是BufferedMutator使用的典型场景。MapReduce作业需要批量写入，但是无法找到恰当的点执行flush。BufferedMutator接收MapReduce作业发送来的Put数据后，会根据某些因素（比如接收的Put数据的总量）启发式地执行Batch Put操作，且会异步的提交Batch Put请求，这样MapReduce作业的执行也不会被打断。
BufferedMutator也可以用在一些特殊的情况上。MapReduce作业的每个线程将会拥有一个独立的BufferedMutator对象。一个独立的BufferedMutator也可以用在大容量的在线系统上来执行批量Put操作，但是这时需要注意一些极端情况比如JVM异常或机器故障，此时有可能造成数据丢失。
如下是几个使用BufferedMutator的实例。
实例一：
protected void instantiateHTable() throws IOException {
for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) {
BufferedMutatorParams params = new BufferedMutatorParams(getTableName(i));
params.writeBufferSize(4 * 1024 * 1024);
BufferedMutator table = connection.getBufferedMutator(params);
this.tables[i] = table;
}
}
这段代码演示了创建BufferedMutator的一般过程。如前文所说，BufferedMutator的配置通常通过BufferedMutatorParams完成。获取BufferedMutator实例则是通过Connection对象。在这里还调整了一个writeBuffer的设置，这里等于是覆盖了HBase配置文件中的“hbase.client.write.buffer”设置。
实例二：
public int run(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
final BufferedMutator.ExceptionListener listener = new BufferedMutator.ExceptionListener() {
@Override
public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator mutator) {
for (int i = 0; i < e.getNumExceptions(); i++) {
LOG.info("Failed to sent put " + e.getRow(i) + ".");
}
}
};
BufferedMutatorParams params = new BufferedMutatorParams(TABLE).listener(listener);
try (final Connection conn = ConnectionFactory.createConnection(getConf()); final BufferedMutator mutator = conn.getBufferedMutator(params)) {
final ExecutorService workerPool = Executors.newFixedThreadPool(POOL_SIZE);
List<Future<Void>> futures = new ArrayList<>(TASK_COUNT);
for (int i = 0; i < TASK_COUNT; i++) {
futures.add(workerPool.submit(new Callable<Void>() {
@Override
public Void call() throws Exception {
Put p = new Put(Bytes.toBytes("someRow"));
p.addColumn(FAMILY, Bytes.toBytes("someQualifier"), Bytes.toBytes("some value"));
mutator.mutate(p);
return null;
}
}));
}
for (Future<Void> f : futures) {
f.get(5, TimeUnit.MINUTES);
}
workerPool.shutdown();
} catch (IOException e) {
LOG.info("exception while creating/destroying Connection or BufferedMutator", e);
}
return 0;
}
这段代码只看前半部分即可。与实例一的不同在于这里新建了一个ExceptionListener接口实现，并添加到BufferedMutatorParams中替换了BufferedMutatorParams的默认ExceptionListener实现。
实例三：
public void testBufferSizeFlush() throws Exception {
int maxSize = 1024;
BufferedMutatorParams params = new BufferedMutatorParams(TABLE_NAME).writeBufferSize(maxSize);
try (BufferedMutator mutator = getConnection().getBufferedMutator(params)) {
Assert.assertTrue(0 == mutator.getWriteBufferSize() || maxSize == mutator.getWriteBufferSize());
Put put = getPut();
mutator.mutate(put);
Assert.assertTrue(mutator.getWriteBufferSize() > 0);
Put largePut = new Put(dataHelper.randomData("testrow-"));
largePut.addColumn(COLUMN_FAMILY, qualifier, Bytes.toBytes(RandomStringUtils.randomAlphanumeric(maxSize * 2)));
long heapSize = largePut.heapSize();
Assert.assertTrue("largePut heapsize is : " + heapSize, heapSize > maxSize);
mutator.mutate(largePut);
Assert.assertTrue(0 == mutator.getWriteBufferSize() || maxSize == mutator.getWriteBufferSize());
}
}
这里是一段BufferedMutator测试代码。
实例四：
public void setup(Context context) throws IOException {
conf = context.getConfiguration();
recordsToWrite = conf.getLong(NUM_TO_WRITE_KEY, NUM_TO_WRITE_DEFAULT);
String tableName = conf.get(TABLE_NAME_KEY, TABLE_NAME_DEFAULT);
numBackReferencesPerRow = conf.getInt(NUM_BACKREFS_KEY, NUM_BACKREFS_DEFAULT);
this.connection = ConnectionFactory.createConnection(conf);
mutator = connection.getBufferedMutator(new BufferedMutatorParams(TableName.valueOf(tableName)).writeBufferSize(4 * 1024 * 1024));
String taskId = conf.get("mapreduce.task.attempt.id");
Matcher matcher = Pattern.compile(".+_m_(\\d+_\\d+)").matcher(taskId);
if (!matcher.matches()) {
throw new RuntimeException("Strange task ID: " + taskId);
}
shortTaskId = matcher.group(1);
rowsWritten = context.getCounter(Counters.ROWS_WRITTEN);
refsWritten = context.getCounter(Counters.REFERENCES_WRITTEN);
}
实例五：
protected void setup(Context context) throws IOException, InterruptedException {
this.conf = context.getConfiguration();
Connection c = ConnectionFactory.createConnection(this.conf);
this.fs = FileSystem.get(conf);
mobCompactionDelay = conf.getLong(SweepJob.MOB_SWEEP_JOB_DELAY, SweepJob.ONE_DAY);
String tableName = conf.get(TableInputFormat.INPUT_TABLE);
String familyName = conf.get(TableInputFormat.SCAN_COLUMN_FAMILY);
TableName tn = TableName.valueOf(tableName);
this.familyDir = MobUtils.getMobFamilyPath(conf, tn, familyName);
Admin admin = c.getAdmin();
try {
family = admin.getTableDescriptor(tn).getFamily(Bytes.toBytes(familyName));
if (family == null) {
throw new InvalidFamilyOperationException("Column family '" + familyName + "' does not exist. It might be removed.");
}
} finally {
try {
admin.close();
} catch (IOException e) {
LOG.warn("Failed to close the HBaseAdmin", e);
}
}
Configuration copyOfConf = new Configuration(conf);
copyOfConf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0f);
this.cacheConfig = new CacheConfig(copyOfConf);
table = c.getBufferedMutator(new BufferedMutatorParams(tn).writeBufferSize(1 * 1024 * 1024));
memstore = new MemStoreWrapper(context, fs, table, family, new DefaultMemStore(), cacheConfig);
this.compactionBegin = conf.getLong(MobConstants.MOB_SWEEP_TOOL_COMPACTION_START_DATE, 0);
mobTableDir = FSUtils.getTableDir(MobUtils.getMobHome(conf), tn);
}
如有必要，稍后抽时间对代码做些简单说明。
######
发表评论