大数据应用之HBase数据插入性能优化之多线程并行插入测试案例

  1 import org.apache.hadoop.conf.Configuration;
  2 import org.apache.hadoop.hbase.HBaseConfiguration;
  3 import java.io.BufferedReader;
  4 import java.io.File;
  5 import java.io.FileNotFoundException;
  6 import java.io.FileReader;
  7 import java.io.IOException;
  8 import java.util.ArrayList;
  9 import java.util.List;
 10 import java.util.Random;
 11 
 12 import org.apache.hadoop.conf.Configuration;
 13 import org.apache.hadoop.hbase.HBaseConfiguration;
 14 import org.apache.hadoop.hbase.client.HBaseAdmin;
 15 import org.apache.hadoop.hbase.client.HTable;
 16 import org.apache.hadoop.hbase.client.HTableInterface;
 17 import org.apache.hadoop.hbase.client.HTablePool;
 18 import org.apache.hadoop.hbase.client.Put;
 19 
 20 public class HBaseImportEx {
 21     static Configuration hbaseConfig = null;
 22     public static HTablePool pool = null;
 23     public static String tableName = "T_TEST_1";
 24     static{
 25          //conf = HBaseConfiguration.create();
 26          Configuration HBASE_CONFIG = new Configuration();
 27          HBASE_CONFIG.set("hbase.master", "192.168.230.133:60000");
 28          HBASE_CONFIG.set("hbase.zookeeper.quorum", "192.168.230.133");
 29          HBASE_CONFIG.set("hbase.zookeeper.property.clientPort", "2181");
 30          hbaseConfig = HBaseConfiguration.create(HBASE_CONFIG);
 31          
 32          pool = new HTablePool(hbaseConfig, 1000); 
 33     }
 34     /*
 35      * Insert Test single thread
 36      * */
 37     public static void SingleThreadInsert()throws IOException
 38     {
 39         System.out.println("---------开始SingleThreadInsert测试----------");
 40         long start = System.currentTimeMillis();
 41         //HTableInterface table = null;
 42         HTable table = null;
 43         table = (HTable)pool.getTable(tableName);
 44         table.setAutoFlush(false);
 45         table.setWriteBufferSize(24*1024*1024);
 46         //构造测试数据
 47         List<Put> list = new ArrayList<Put>();
 48         int count = 10000;
 49         byte[] buffer = new byte[350];
 50         Random rand = new Random();
 51         for(int i=0;i<count;i++)
 52         {
 53             Put put = new Put(String.format("row %d",i).getBytes());
 54             rand.nextBytes(buffer);
 55             put.add("f1".getBytes(), null, buffer);
 56             //wal=false
 57             put.setWriteToWAL(false);
 58             list.add(put);    
 59             if(i%10000 == 0)
 60             {
 61                 table.put(list);
 62                 list.clear();    
 63                 table.flushCommits();
 64             }            
 65         }
 66         long stop = System.currentTimeMillis();
 67         //System.out.println("WAL="+wal+",autoFlush="+autoFlush+",buffer="+writeBuffer+",count="+count);
 68           
 69         System.out.println("插入数据："+count+"共耗时："+ (stop - start)*1.0/1000+"s");
 70         
 71         System.out.println("---------结束SingleThreadInsert测试----------");
 72     }
 73     /*
 74      * 多线程环境下线程插入函数 
 75      * 
 76      * */
 77     public static void InsertProcess()throws IOException
 78     {
 79         long start = System.currentTimeMillis();
 80         //HTableInterface table = null;
 81         HTable table = null;
 82         table = (HTable)pool.getTable(tableName);
 83         table.setAutoFlush(false);
 84         table.setWriteBufferSize(24*1024*1024);
 85         //构造测试数据
 86         List<Put> list = new ArrayList<Put>();
 87         int count = 10000;
 88         byte[] buffer = new byte[256];
 89         Random rand = new Random();
 90         for(int i=0;i<count;i++)
 91         {
 92             Put put = new Put(String.format("row %d",i).getBytes());
 93             rand.nextBytes(buffer);
 94             put.add("f1".getBytes(), null, buffer);
 95             //wal=false
 96             put.setWriteToWAL(false);
 97             list.add(put);    
 98             if(i%10000 == 0)
 99             {
100                 table.put(list);
101                 list.clear();    
102                 table.flushCommits();
103             }            
104         }
105         long stop = System.currentTimeMillis();
106         //System.out.println("WAL="+wal+",autoFlush="+autoFlush+",buffer="+writeBuffer+",count="+count);
107           
108         System.out.println("线程:"+Thread.currentThread().getId()+"插入数据："+count+"共耗时："+ (stop - start)*1.0/1000+"s");
109     }
110     
111     
112     /*
113      * Mutil thread insert test
114      * */
115     public static void MultThreadInsert() throws InterruptedException
116     {
117         System.out.println("---------开始MultThreadInsert测试----------");
118         long start = System.currentTimeMillis();
119         int threadNumber = 10;
120         Thread[] threads=new Thread[threadNumber];
121         for(int i=0;i<threads.length;i++)
122         {
123             threads[i]= new ImportThread();
124             threads[i].start();            
125         }
126         for(int j=0;j< threads.length;j++)
127         {
128              (threads[j]).join();
129         }
130         long stop = System.currentTimeMillis();
131           
132         System.out.println("MultThreadInsert："+threadNumber*10000+"共耗时："+ (stop - start)*1.0/1000+"s");        
133         System.out.println("---------结束MultThreadInsert测试----------");
134     }    
135 
136     /**
137      * @param args
138      */
139     public static void main(String[] args)  throws Exception{
140         // TODO Auto-generated method stub
141         //SingleThreadInsert();        
142         MultThreadInsert();
143         
144         
145     }
146     
147     public static class ImportThread extends Thread{
148         public void HandleThread()
149         {                        
150             //this.TableName = "T_TEST_1";
151         
152             
153         }
154         //
155         public void run(){
156             try{
157                 InsertProcess();            
158             }
159             catch(IOException e){
160                 e.printStackTrace();                
161             }finally{
162                 System.gc();
163                 }
164             }            
165         }
166 
167 }