大数据应用之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 }