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的实例。
实例一：
|
1 2 3 4 5 6 7 8 |
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”设置。
实例二：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
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实现。
实例三：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
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测试代码。
实例四：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
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); } |
实例五：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
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); } |
如有必要，稍后抽时间对代码做些简单说明。
######
发表评论