HBase 高性能加入数据 - 按批多“粮仓”式解决办法

摘要:如何从HBase中的海量数据中,以很快的速度的获取大批量数据,这一议题已经在《HBase 高性能获取数据》(http://www.cnblogs.com/wgp13x/p/4245182.html)一文中给出了解决办法。那么,如何向HBase中高性能的插入数据呢?经研究表明,光是批量写入也还是不行。网上没有现成的方法。本文针对这一问题,给出了一个解决方案。它采用了多线程按批“多粮仓”的方式,经过验证,能较好的达到高速度的效果。

关键词:hbase, 高性能, 多线​程, 算法

解决问题:如何向HBase中高性能的插入数据。


Solr和HBase专辑

1、“关于Solr的使用总结的心得体会”(http://www.cnblogs.com/wgp13x/p/3742653.html)

2、“中文分词器性能比较​”(http://www.cnblogs.com/wgp13x/p/3748764.html)

3、“Solr与HBase架构设计”(http://www.cnblogs.com/wgp13x/p/a8bb8ccd469c96917652201007ad3c50.html)

4、 “大数据架构: 使用HBase和Solr将存储与索引放在不同的机器上”(http://www.cnblogs.com/wgp13x/p/3927979.html)

5、“一个自定义 HBase Filter -通过RowKeys来高性能获取数据”(http://www.cnblogs.com/wgp13x/p/4196466.html)

6、“HBase 高性能获取数据 - 多线程批量式解决办法”(http://www.cnblogs.com/wgp13x/p/4245182.html


在实现向海量数据库(HBase + Solr)插入数据的过程中,发现存在这么一个问题:如果要插入的数据是一条一条传入的,每秒还能传入上千条那么多,那么入库的速度非常慢,以每秒数条记,此时面临着读写问题,读的速度远远跟不上写的速度,会造成大量数据积累,严重的会造成磁盘写爆掉。首先要说明的是,我们是实现了“按批入库”的接口的:public int importDataBatch(String dataTypeCode, List<Map<String, String>> datas),既然有这一接口,那么为什么还会写入速度如此慢呢?HBase 高性能加入数据 - 按批多“粮仓”式解决办法我们若是只对此接口进行性能测试的话,可以发现传入的批量数据的大小对性能的影响非常大,像这种一条一条调用importDataBatch的,简直是暴殄天物;当然,我们希望一批数据里有成千上万条数据一起提交入库才过瘾,如果配置的好的话,像《HBase 高性能获取数据》一文中“HBase基本说明与性能测试”所叙述的那样:“单台机器能够实现1w/s~3w/s之间的插入速度”。那么,对于这种实际情况中一条一条的数据插入,我们应该如何处理呢?

解决办法其实很简单,那就是“广积粮,深挖仓”HBase 高性能加入数据 - 按批多“粮仓”式解决办法。首先,它的传入速度还算快,能达到每秒上千条;其次,传入的数据是一条一条组织的;如果遇到每条数据都可能对应不同的数据类型,那会更麻烦。“广积粮,深挖仓”也得讲究方法,仓该如何组织,小米和大豆是否应该存放于不同的仓库里,积粮久了会不会发霉,粮应积多久就要吃掉了,这都是问题。我的方案是:“对于不同的数据类型粮,建造不同的粮仓;一旦粮仓满了,要立刻打开粮仓分了吃掉;对于积攒超过5秒的而没被吃掉的粮,要立即吃掉”,这样就保证了数据不会丢失,并且处理速度超快。

下面是这一算法的具体代码实现:

 
/**
 * 数据处理
 * 
 * @author wanganqi
 * @version v1.0
 * @since 2014年8月8日上午10:11:07

*/

public class DataProcessHandler
{
    private static final int MULIPLE_CONT = 10;    // 10个多粮仓
    private Random random = new Random();

private MultipleImportDatas[] multipleImportDatas = new MultipleImportDatas[MULIPLE_CONT];    // 10个多粮仓入库

    public DataProcessHandler()
    {
        for (int i = 0; i < MULIPLE_CONT; i++)
        {
            multipleImportDatas[i] = new MultipleImportDatas();
        }

}

    private void 处理来的每一条数据(String dataTypeCode, List<Map<String, String>> datas)
    {
        int index = random.nextInt(MULIPLE_CONT);    // 随机选择1个多粮仓入库
        multipleImportDatas[index].concurrentImportDatas(dataTypeCode, datas, false);    // 线程安全入数据。入库需满仓
    } 
 
 
public class MultipleImportDatas    // 多粮仓线程安全入库
{
    private ConcurrentHashMap<String, List<Map<String, String>>> inputBuffer = new ConcurrentHashMap<String, List<Map<String, String>>>();    // 各粮仓,按数据类型划分
    private ConcurrentHashMap<String, Date> inputBufferDate = new ConcurrentHashMap<String, Date>();    // 各粮仓,最近清仓时间
    private final int BUFFER_SIZE = 1000;    // 各粮仓大小
    private final int BUFFER_MONITOR_TIME = 5000;    // 对粮仓监控间隔
    private Thread inputBufferMonitor = new InputBufferMonitor();    // 对粮仓进行监控
 
    public MultipleImportDatas()
    {
        inputBufferMonitor.start();
    }
 
    class InputBufferMonitor extends Thread
    {
        public void run()
        {
            while (true)
            {
                try
                {
                    Thread.sleep(BUFFER_MONITOR_TIME);
                }
                catch (InterruptedException e1)
                {
                    e1.printStackTrace();
                }
                for (Map.Entry<String, List<Map<String, String>>> entity : inputBuffer
                    .entrySet())   // 各粮仓监管
                {
                    String dataType = entity.getKey();
                    concurrentImportDatas(dataType, null, true);    // 线程安全入数据。入库需时间到
                }
            }
        }
    }
 
    private void resetInputBufferDate(String dateTypeCode)   // 更新清仓时间
    {
        Date date = new Date();
        if (!inputBufferDate.containsKey(dateTypeCode))
        {
            inputBufferDate.put(dateTypeCode, date);
        }
        else
        {
            inputBufferDate.replace(dateTypeCode, date);
        }
    }
 
    public void concurrentImportDatas(String dataTypeCode,
        List<Map<String, String>> datas, boolean timeUP)    // 线程安全入数据。入库需满仓或时间到
    {
        DataHBaseSolrBLL dataHBaseSolrBLL = (DataHBaseSolrBLL) SpringRabbitMQSupport
            .getContext().getBean("dataHBaseSolrBLLImpl");
        if (datas != null && datas.size() >= BUFFER_SIZE)    // 一次入粮多
        {
            try
            {
                dataHBaseSolrBLL.importDataBatch(dataTypeCode, datas);   // 批量入库
            }
            catch (DatabaseException | BusinessException e)
            {
                e.printStackTrace();
            }
            return;
        }
        synchronized (inputBuffer)
        {
            if (!inputBuffer.containsKey(dataTypeCode))     // 无仓
            {
                if (datas != null)
                {
                    inputBuffer.put(dataTypeCode, datas);   // 建仓
                }
                return;
            }
            if (timeUP)   // 时间到
            {
                Date date = new Date();
                if (((!inputBufferDate.containsKey(dataTypeCode)) || (inputBufferDate
                    .containsKey(dataTypeCode) && date.getTime()
                    - inputBufferDate.get(dataTypeCode).getTime() > BUFFER_MONITOR_TIME))
                    && inputBuffer.get(dataTypeCode).size() > 0)
                {
                    importInputBufferAndClear(dataTypeCode);   // 清仓
                }
            }
            else
            {
                List<Map<String, String>> ldatas = inputBuffer
                    .get(dataTypeCode);
                if (datas.size() + ldatas.size() > BUFFER_SIZE)   // 要满仓
                {
                    importInputBufferAndClear(dataTypeCode);   // 清仓
                }
                ldatas.addAll(datas);    // 未满仓,入仓
            }
        }
    }
 
    private void importInputBufferAndClear(String dataTypeCode)   // 清仓
    {
        DataHBaseSolrBLL dataHBaseSolrBLL = (DataHBaseSolrBLL) SpringRabbitMQSupport
            .getContext().getBean("dataHBaseSolrBLLImpl");
        List<Map<String, String>> ldatas = inputBuffer.get(dataTypeCode);
        try
        {
            dataHBaseSolrBLL.importDataBatch(dataTypeCode, ldatas);   // 批量入库
        }
        catch (DatabaseException | BusinessException e)
        {
            e.printStackTrace();
        }
        ldatas.clear();
        resetInputBufferDate(dataTypeCode);   // 更新清仓时间
    }

}

说明:这里起了多个“多粮仓”来进行入库,是为了避免线程同步造成的性能降低。测试表明,如果只起一个“多粮仓”来进行入库,每秒可以入库300多条数据,但若是起10个“多粮仓”来进行入库,每秒可以入库1200多条数据,这说明线程读写同步造成的时延现象还是蛮严重的。说明一下:这里的测试环境与《HBase 高性能获取数据》一文中“HBase基本说明与性能测试”环境不同,这里的加入了数据验证等过程,因此与其所描述的不具备可比性。

明天就是周末咯,偶和亲爱的老婆各请了15天的婚假,过几天就要离开这寒冷的冬季,去夏季巴厘岛度蜜月咯,好开心HBase 高性能加入数据 - 按批多“粮仓”式解决办法

上一篇:7 Hbase put方式插入数据


下一篇:【Android群英传】学习笔记(二)