关于hbase多线程写入问题

Question

@cloud-coder 你好，想跟你请教个问题：

多线程写入hbase，官方已经摒弃了HtablePool方式写入，推荐使用HConnection#getTable方式。

问题：我不想put一次，就flushCommit。于是我在client端使用了setFlusthAo(false)以及缓存的大小XMB。

这种客户端只能达到table达到XMB的时候才能自动flushCommit。但是我们的业务只允许几秒延迟的，于是乎我使用了每隔三秒flushCommit一次（在getTable的时候，启动线程对这个table每隔3s flushCommit一次）。

发现Htable线程不是安全的，在我flushCommit的同时，会有其他线程对这个table进行写入。然后报异常

**ModificationException （具体名字忘记了）

请问对于这种问题，你有没有碰到过？然后是怎么处理的，谢谢！

Answer 1

当然是以我自身的业务来说

Answer 2

hbase region的heap设置大点，如12G

使用的是什么版本的hbase,CDH5?

Answer 3

@cloud-coder  感谢您的答复。

我这边还有几个问题：

1、压缩完后（数据量达到1.6T），读的速度立马上升，但是写入速度速度下降，运行一段时间，会出现rpc连接一直断开现象。目前无法下手怎么优化，请问下你们对这方面怎么优化方法？

（我们有5台regionserver，压缩完后每台有70个region，每region 就1~2 个Store， 最大StoreFile是10G，只有一个簇）。

Answer 4

你的问题我没有遇到过，

Answer 5

可以在方法中通过调用table.flushcommits进行提交，一般情况下一个事务提交一次

可以是一行或多行数据

Answer 6

我在评论中回复了，请帮忙看下。谢谢:)

Answer 7

// 初始化UserProvider
        UserProvider userProvider = UserProvider.instantiate(conf);

        // 初始化admin,ConnectionCache
        try {
            connectionCache = new ConnectionCache(conf, userProvider, cleanInterval, maxIdleTime);
            admin = new HBaseAdmin(conf);
        }
        catch (Exception e) {
            LOG.error("创建HBaseAdmin失败:" + e.getMessage());
        }

        // 初始化HTableFactory
        tableFactory = new HTableFactory() {
            @Override
            public HTableInterface createHTableInterface(Configuration config, byte[] tableName) {
                try {
                    return connectionCache.getTable(Bytes.toString(tableName));
                }
                catch (IOException ioe) {
                    throw new RuntimeException(ioe);
                }
            }
        };

        // 初始化Cache
        htablePools =
                CacheBuilder.newBuilder().expireAfterAccess(maxIdleTime, TimeUnit.MILLISECONDS).softValues()
                    .concurrencyLevel(4).build();
        htablePoolCreater = new Callable<HTablePool>() {
            public HTablePool call() {
                return new HTablePool(conf, maxPoolSize, tableFactory);
            }
        };


/**
     * 获取HTableInterface实例
     * 
     * @param tableName
     * @return
     */
    private static HTableInterface getTable(String tableName) {
        String currentUser = connectionCache.getEffectiveUser();
        try {
            HTablePool htablePool = htablePools.get(currentUser, htablePoolCreater);
            return htablePool.getTable(tableName);
        }
        catch (ExecutionException ee) {
            throw new RuntimeException(ee);
        }
    }