大數據應用之HBase數據插入性能優化之多線程並行插入測試案例

一、引言：

　　上篇文章提起關於HBase插入性能優化設計到的五個參數，從參數配置的角度給大家提供了一個性能測試環境的實驗代碼。根據網友的反饋，基於單線程的模式實現的數據插入畢竟有限。通過個人實測，在我的虛擬機環境下，單線程插入數據的值約為4w/s。集群指標是：CPU雙核1.83，虛擬機512M內存，集群部署單點模式。本文給出了基於多線程並發模式的，測試代碼案例和實測結果，希望能給大家一些啟示：

二、源程序：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;

public class HBaseImportEx {
    static Configuration hbaseConfig = null;
    public static HTablePool pool = null;
    public static String tableName = "T_TEST_1";
    static{
         //conf = HBaseConfiguration.create();
         Configuration HBASE_CONFIG = new Configuration();
         HBASE_CONFIG.set("hbase.master", "192.168.230.133:60000");
         HBASE_CONFIG.set("hbase.zookeeper.quorum", "192.168.230.133");
         HBASE_CONFIG.set("hbase.zookeeper.property.clientPort", "2181");
         hbaseConfig = HBaseConfiguration.create(HBASE_CONFIG);
         
         pool = new HTablePool(hbaseConfig, 1000); 
    }
    /*
     * Insert Test single thread
     * */
    public static void SingleThreadInsert()throws IOException
    {
        System.out.println("---------開始SingleThreadInsert測試----------");
        long start = System.currentTimeMillis();
        //HTableInterface table = null;
        HTable table = null;
        table = (HTable)pool.getTable(tableName);
        table.setAutoFlush(false);
        table.setWriteBufferSize(24*1024*1024);
        //構造測試數據
        List<Put> list = new ArrayList<Put>();
        int count = 10000;
        byte[] buffer = new byte[350];
        Random rand = new Random();
        for(int i=0;i<count;i++)
        {
            Put put = new Put(String.format("row %d",i).getBytes());
            rand.nextBytes(buffer);
            put.add("f1".getBytes(), null, buffer);
            //wal=false
            put.setWriteToWAL(false);
            list.add(put);    
            if(i%10000 == 0)
            {
                table.put(list);
                list.clear();    
                table.flushCommits();
            }            
        }
        long stop = System.currentTimeMillis();
        //System.out.println("WAL="+wal+",autoFlush="+autoFlush+",buffer="+writeBuffer+",count="+count);
          
        System.out.println("插入數據："+count+"共耗時："+ (stop - start)*1.0/1000+"s");
        
        System.out.println("---------結束SingleThreadInsert測試----------");
    }
    /*
     * 多線程環境下線程插入函數 
     * 
     * */
    public static void InsertProcess()throws IOException
    {
        long start = System.currentTimeMillis();
        //HTableInterface table = null;
        HTable table = null;
        table = (HTable)pool.getTable(tableName);
        table.setAutoFlush(false);
        table.setWriteBufferSize(24*1024*1024);
        //構造測試數據
        List<Put> list = new ArrayList<Put>();
        int count = 10000;
        byte[] buffer = new byte[256];
        Random rand = new Random();
        for(int i=0;i<count;i++)
        {
            Put put = new Put(String.format("row %d",i).getBytes());
            rand.nextBytes(buffer);
            put.add("f1".getBytes(), null, buffer);
            //wal=false
            put.setWriteToWAL(false);
            list.add(put);    
            if(i%10000 == 0)
            {
                table.put(list);
                list.clear();    
                table.flushCommits();
            }            
        }
        long stop = System.currentTimeMillis();
        //System.out.println("WAL="+wal+",autoFlush="+autoFlush+",buffer="+writeBuffer+",count="+count);
          
        System.out.println("線程:"+Thread.currentThread().getId()+"插入數據："+count+"共耗時："+ (stop - start)*1.0/1000+"s");
    }
    
    
    /*
     * Mutil thread insert test
     * */
    public static void MultThreadInsert() throws InterruptedException
    {
        System.out.println("---------開始MultThreadInsert測試----------");
        long start = System.currentTimeMillis();
        int threadNumber = 10;
        Thread[] threads=new Thread[threadNumber];
        for(int i=0;i<threads.length;i++)
        {
            threads[i]= new ImportThread();
            threads[i].start();            
        }
        for(int j=0;j< threads.length;j++)
        {
             (threads[j]).join();
        }
        long stop = System.currentTimeMillis();
          
        System.out.println("MultThreadInsert："+threadNumber*10000+"共耗時："+ (stop - start)*1.0/1000+"s");        
        System.out.println("---------結束MultThreadInsert測試----------");
    }    

    /**
     * @param args
     */
    public static void main(String[] args)  throws Exception{
        // TODO Auto-generated method stub
        //SingleThreadInsert();        
        MultThreadInsert();
        
        
    }
    
    public static class ImportThread extends Thread{
        public void HandleThread()
        {                        
            //this.TableName = "T_TEST_1";
        
            
        }
        //
        public void run(){
            try{
                InsertProcess();            
            }
            catch(IOException e){
                e.printStackTrace();                
            }finally{
                System.gc();
                }
            }            
        }

}

三、說明

1.線程數設置需要根據本集群硬件參數，實際測試得出。否則線程過多的情況下，總耗時反而是下降的。

2.單筆提交數對性能的影響非常明顯，需要在自己的環境下，找到最理想的數值，這個需要與單條記錄的字節數相關。

四、測試結果

---------開始MultThreadInsert測試----------

線程:8插入數據：10000共耗時：1.328s
線程:16插入數據：10000共耗時：1.562s
線程:11插入數據：10000共耗時：1.562s
線程:10插入數據：10000共耗時：1.812s
線程:13插入數據：10000共耗時：2.0s
線程:17插入數據：10000共耗時：2.14s
線程:14插入數據：10000共耗時：2.265s
線程:9插入數據：10000共耗時：2.468s
線程:15插入數據：10000共耗時：2.562s
線程:12插入數據：10000共耗時：2.671s
MultThreadInsert：100000共耗時：2.703s
---------結束MultThreadInsert測試----------

 備注：該技術專題討論正在群Hadoop高級交流群：293503507同步直播中，敬請關注。