hbase持有者工具类

  1 import org.apache.commons.lang.StringUtils;
  2 import org.apache.hadoop.conf.Configuration;
  3 import org.apache.hadoop.hbase.*;
  4 import org.apache.hadoop.hbase.client.*;
  5 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
  6 import org.apache.hadoop.hbase.util.Bytes;
  7 import org.slf4j.Logger;
  8 import org.slf4j.LoggerFactory;
  9 
 10 import java.io.Closeable;
 11 import java.io.IOException;
 12 import java.io.Serializable;
 13 import java.util.ArrayList;
 14 import java.util.HashMap;
 15 import java.util.List;
 16 import java.util.Map;
 17 
 18 /**
 19  * <p>
 20  * HBase工具类
 21  * </p>
 22  *
 23  * @author 用户名 2015年6月18日 上午8:58:38
 24  * @version V1.0
 25  * @modify by user: {修改人} 2015年6月18日
 26  * @modify by reason:{方法名}:{原因}
 27  */
 28 public class HBaseHolder implements Serializable,Closeable {
 29 
 30     // 日志记录器
 31     protected static final Logger LOGGER = LoggerFactory.getLogger(HBaseHolder.class);
 32     private static final int DEFAULT_MAX_VERSIONS = 3;
 33     // HBase配置
 34     private Configuration config;
 35     private Admin admin;
 36     private Connection connection;
 37 
 38     public HBaseHolder() {
 39         config = HBaseConfiguration.create();
 40     }
 41 
 42     public HBaseHolder(Configuration config) {
 43         this.config = config;
 44     }
 45 
 46     /**
 47      * 从连接池获取HTable对象
 48      *
 49      * @param tableName
 50      * @return
 51      * @throws IOException
 52      * @author
 53      */
 54     public Table getTable(String tableName) throws IOException {
 55         return getConnection().getTable(TableName.valueOf(tableName));
 56     }
 57 
 58     /**
 59      * 获取HAdmin对象，建表等操作
 60      *
 61      * @return
 62      * @throws IOException
 63      * @author 
 64      */
 65     public Admin getHBaseAdmin() throws IOException {
 66         if (admin == null) {
 67             admin = getConnection().getAdmin();
 68         }
 69         return admin;
 70     }
 71 
 72     /**
 73      * 关闭HTable对象
 74      *
 75      * @param table
 76      * @author
 77      */
 78     public void doCloseTable(Table table) {
 79         if (table == null) {
 80             return;
 81         }
 82         try {
 83             table.close();
 84         } catch (IOException e) {
 85             e.printStackTrace();
 86         }
 87     }
 88 
 89     /**
 90      * 创建表操作
 91      *
 92      * @param tableName
 93      * @param families
 94      * @author
 95      */
 96     public void createTable(String tableName, String[] families) {
 97         createTable(tableName, DEFAULT_MAX_VERSIONS, null, families);
 98     }
 99 
100     /**
101      * 创建表操作
102      *
103      * @param tableName
104      * @param splitKeys
105      * @param families
106      * @author
107      */
108     public void createTable(String tableName, byte[][] splitKeys, String[] families) {
109         createTable(tableName, DEFAULT_MAX_VERSIONS, splitKeys, families);
110     }
111 
112     /**
113      * 创建表操作
114      *
115      * @param tableName
116      * @param maxVersions
117      * @param families
118      * @author
119      */
120     public void createTable(String tableName, int maxVersions, String[] families) {
121         createTable(tableName, maxVersions, null, families);
122     }
123 
124     /**
125      * 创建表操作
126      *
127      * @param tableName
128      * @param family
129      * @author
130      */
131     public void createTable(String tableName, int maxVersions, byte[][] splitKeys, String[] families) {
132         // 参数判空
133         if (StringUtils.isBlank(tableName) || families == null || families.length <= 0) {
134             return;
135         }
136         try {
137             // 表不存在则创建
138             if (!getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
139                 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
140                 for (String family : families) {
141                     HColumnDescriptor columnDescriptor = new HColumnDescriptor(family);
142                     columnDescriptor.setCompressionType(Algorithm.SNAPPY);
143                     columnDescriptor.setMaxVersions(maxVersions);
144                     desc.addFamily(columnDescriptor);
145                 }
146                 if (splitKeys != null) {
147                     getHBaseAdmin().createTable(desc, splitKeys);
148                 } else {
149                     getHBaseAdmin().createTable(desc);
150                 }
151             } else {
152                 LOGGER.warn("Table " + tableName + " already exists.");
153             }
154         } catch (IOException e) {
155             LOGGER.error("", e);
156         }
157     }
158 
159     /**
160      * 删除表
161      *
162      * @param tableName
163      * @author
164      */
165     public void dropTable(String tableName) {
166         Admin admin = null;
167         try {
168             admin = getHBaseAdmin();
169             if (admin.tableExists(TableName.valueOf(tableName))) {
170                 admin.disableTable(TableName.valueOf(tableName));
171                 admin.deleteTable(TableName.valueOf(tableName));
172             }
173         } catch (IOException e) {
174             LOGGER.error("drop table error." + e);
175         } finally {
176             if (null != admin) {
177                 try {
178                     admin.close();
179                 } catch (IOException e) {
180                     LOGGER.error("close admin error " + e);
181                 }
182             }
183         }
184     }
185 
186     /**
187      * 获取单个列值
188      *
189      * @param tableName
190      * @param rowkey
191      * @return
192      * @author
193      */
194     public byte[] get(String tableName, String rowkey, String family, String qualifier) {
195         Table table = null;
196         try {
197             Get get = new Get(Bytes.toBytes(rowkey));
198             get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
199             table = getTable(tableName);
200             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
201                 Result result = table.get(get);
202                 return result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
203             } else {
204                 LOGGER.warn("Table " + tableName + " does not exist.");
205             }
206         } catch (IOException e) {
207             LOGGER.error("获取列值失败！ " + e);
208             e.printStackTrace();
209         }
210         return null;
211     }
212 
213     /**
214      * 获取单个列值，字符串返回
215      *
216      * @param tableName
217      * @param rowkey
218      * @return
219      * @author zhanglei11
220      */
221     public String getString(String tableName, String rowkey, String family, String qualifier) {
222         return Bytes.toString(get(tableName, rowkey, family, qualifier));
223     }
224 
225     /**
226      * 获取一行中某列族的值
227      *
228      * @param tableName
229      * @param rowkey
230      * @return
231      * @author
232      */
233     public Map<String, byte[]> getMapByKeyAndFamily(String tableName, String rowkey, String family) {
234         Map<String, byte[]> map = new HashMap<String, byte[]>();
235         Table table = null;
236         try {
237             Get get = new Get(Bytes.toBytes(rowkey));
238             get.addFamily(Bytes.toBytes(family));
239             table = getTable(tableName);
240             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
241                 Result result = table.get(get);
242                 for (Cell cell : result.rawCells()) {
243                     byte[] q = CellUtil.cloneQualifier(cell);
244                     byte[] v = CellUtil.cloneValue(cell);
245                     map.put(Bytes.toString(q), v);
246                 }
247             } else {
248                 LOGGER.warn("Table " + tableName + " does not exist.");
249             }
250         } catch (IOException e) {
251             e.printStackTrace();
252         }
253         return map;
254     }
255 
256     /**
257      * 获取一整行的值
258      *
259      * @param tableName
260      * @param rowkey
261      * @return
262      * @author
263      */
264     public Map<String, byte[]> getRowMap(String tableName, String rowkey) {
265         Map<String, byte[]> map = new HashMap<String, byte[]>();
266         Table table = null;
267         try {
268             Get get = new Get(Bytes.toBytes(rowkey));
269             table = getTable(tableName);
270             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
271                 Result result = table.get(get);
272                 for (Cell cell : result.rawCells()) {
273                     byte[] q = CellUtil.cloneQualifier(cell);
274                     byte[] v = CellUtil.cloneValue(cell);
275                     map.put(Bytes.toString(q), v);
276                 }
277             } else {
278                 LOGGER.warn("Table " + tableName + " does not exist.");
279             }
280         } catch (IOException e) {
281             e.printStackTrace();
282         }
283         return map;
284     }
285 
286     /**
287      * 获取记录
288      *
289      * @param tableName
290      * @param rowkeys
291      * @return
292      */
293     public Result[] getRecodes(String tableName, List<String> rowkeys) {
294         Table table = null;
295         if (rowkeys == null || rowkeys.size() == 0) {
296             LOGGER.warn("Has no rowkeys to get.");
297             return null;
298         }
299         try {
300             List<Get> gets = new ArrayList<>();
301             Get get = null;
302             for (String rowkey : rowkeys) {
303                 get = new Get(Bytes.toBytes(rowkey));
304                 gets.add(get);
305             }
306             table = getTable(tableName);
307             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
308                 Result[] results = table.get(gets);
309                 return results;
310             } else {
311                 LOGGER.warn("Table " + tableName + " does not exist.");
312                 return null;
313             }
314         } catch (IOException e) {
315             LOGGER.error("get table [{}] recodes error", tableName, e);
316             return null;
317         }
318     }
319 
320     /**
321      * 获取记录
322      *
323      * @param tableName
324      * @param rowkey
325      * @return
326      */
327     public Result getRecode(String tableName, String rowkey) {
328         Table table = null;
329         try {
330             Get get = new Get(Bytes.toBytes(rowkey));
331             table = getTable(tableName);
332             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
333                 Result result = table.get(get);
334                 return result;
335             } else {
336                 LOGGER.warn("Table " + tableName + " does not exist.");
337                 return null;
338             }
339         } catch (IOException e) {
340             LOGGER.error("get table [{}] recodes error", tableName, e);
341             return null;
342         }
343     }
344 
345     /**
346      * 获取记录
347      *
348      * @param tableName
349      * @param gets
350      * @return
351      */
352     public Result[] getRecodesByGets(String tableName, List<Get> gets) {
353         Table table = null;
354         if (gets == null || gets.size() == 0) {
355             LOGGER.warn("Has no gets to get.");
356             return null;
357         }
358         try {
359             table = getTable(tableName);
360             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
361                 Result[] results = table.get(gets);
362                 return results;
363             } else {
364                 LOGGER.warn("Table " + tableName + " does not exist.");
365                 return null;
366             }
367         } catch (IOException e) {
368             LOGGER.error("get table [{}] recodes error", tableName, e);
369             return null;
370         }
371     }
372 
373     /**
374      * 获取记录
375      *
376      * @param tableName
377      * @param get
378      * @return
379      */
380     public Result getRecodeByGet(String tableName, Get get) {
381         Table table = null;
382         try {
383             table = getTable(tableName);
384             if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
385                 Result result = table.get(get);
386                 return result;
387             } else {
388                 LOGGER.warn("Table " + tableName + " does not exist.");
389                 return null;
390             }
391         } catch (IOException e) {
392             LOGGER.error("get table [{}] recodes error", tableName, e);
393             return null;
394         }
395     }
396 
397     /**
398      * 检测HBase服务是否可用
399      *
400      * @return
401      * @author
402      */
403     public boolean isHBaseAvailable() {
404         try {
405             HBaseAdmin.checkHBaseAvailable(config);
406         } catch (ZooKeeperConnectionException zkce) {
407             LOGGER.error("", zkce);
408             return false;
409         } catch (MasterNotRunningException e) {
410             LOGGER.error("", e);
411             return false;
412         } catch (Exception e) {
413             LOGGER.error("Check HBase available throws an Exception. We don‘t know whether HBase is running or not.", e);
414             return false;
415         }
416         return true;
417     }
418 
419     public Connection getConnection() {
420         if (connection == null) {
421             try {
422                 connection = ConnectionFactory.createConnection(config);
423             } catch (IOException e) {
424                 LOGGER.error("Create HBase connect error.", e);
425             }
426         }
427         return connection;
428     }
429 
430     @Override
431     public void close() throws IOException {
432         if(admin != null){
433             admin.close();
434             admin = null;
435         }
436         if(connection != null){
437             connection.close();
438             connection = null;
439         }
440     }
441 }