hbase持有者工具类
Posted 金聂政
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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 }
以上是关于hbase持有者工具类的主要内容，如果未能解决你的问题，请参考以下文章
elasticsearch代码片段，及工具类SearchEsUtil.java