HBase-API-ForJava-简介

资料 https://hbase.apache.org/apidocs/index.html http://hbase.apache.org/book.html#external_apis

1、使用Java创建，修改和删除一个表-示例

  1. import java.io.IOException;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.hbase.HBaseConfiguration;
  6. import org.apache.hadoop.hbase.HColumnDescriptor;
  7. import org.apache.hadoop.hbase.HConstants;
  8. import org.apache.hadoop.hbase.HTableDescriptor;
  9. import org.apache.hadoop.hbase.TableName;
  10. import org.apache.hadoop.hbase.client.Admin;
  11. import org.apache.hadoop.hbase.client.Connection;
  12. import org.apache.hadoop.hbase.client.ConnectionFactory;
  13. import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
  15. public class Example {
  17.   private static final String TABLE_NAME = "MY_TABLE_NAME_TOO";
  18.   private static final String CF_DEFAULT = "DEFAULT_COLUMN_FAMILY";
  20.   public static void createOrOverwrite(Admin admin, HTableDescriptor table) throws IOException {
  21.     if (admin.tableExists(table.getTableName())) {
  22.       admin.disableTable(table.getTableName());
  23.       admin.deleteTable(table.getTableName());
  24.     }
  25.     admin.createTable(table);
  26.   }
  28.   public static void createSchemaTables(Configuration config) throws IOException {
  29.     try (Connection connection = ConnectionFactory.createConnection(config);
  30.          Admin admin = connection.getAdmin()) {
  32.       HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
  33.       table.addFamily(new HColumnDescriptor(CF_DEFAULT).setCompressionType(Algorithm.NONE));
  35.       System.out.print("Creating table. ");
  36.       createOrOverwrite(admin, table);
  37.       System.out.println(" Done.");
  38.     }
  39.   }
  41.   public static void modifySchema (Configuration config) throws IOException {
  42.     try (Connection connection = ConnectionFactory.createConnection(config);
  43.          Admin admin = connection.getAdmin()) {
  45.       TableName tableName = TableName.valueOf(TABLE_NAME);
  46.       if (!admin.tableExists(tableName)) {
  47.         System.out.println("Table does not exist.");
  48.         System.exit(-1);
  49.       }
  51.       HTableDescriptor table = admin.getTableDescriptor(tableName);
  53.       // Update existing table
  54.       HColumnDescriptor newColumn = new HColumnDescriptor("NEWCF");
  55.       newColumn.setCompactionCompressionType(Algorithm.GZ);
  56.       newColumn.setMaxVersions(HConstants.ALL_VERSIONS);
  57.       admin.addColumn(tableName, newColumn);
  59.       // Update existing column family
  60.       HColumnDescriptor existingColumn = new HColumnDescriptor(CF_DEFAULT);
  61.       existingColumn.setCompactionCompressionType(Algorithm.GZ);
  62.       existingColumn.setMaxVersions(HConstants.ALL_VERSIONS);
  63.       table.modifyFamily(existingColumn);
  64.       admin.modifyTable(tableName, table);
  66.       // Disable an existing table
  67.       admin.disableTable(tableName);
  69.       // Delete an existing column family
  70.       admin.deleteColumn(tableName, CF_DEFAULT.getBytes("UTF-8"));
  72.       // Delete a table (Need to be disabled first)
  73.       admin.deleteTable(tableName);
  74.     }
  75.   }
  77.   public static void main(String... args) throws IOException {
  78.     Configuration config = HBaseConfiguration.create();
  80.     //Add any necessary configuration files (hbase-site.xml, core-site.xml)
  81.     config.addResource(new Path(System.getenv("HBASE_CONF_DIR"), "hbase-site.xml"));
  82.     config.addResource(new Path(System.getenv("HADOOP_CONF_DIR"), "core-site.xml"));
  83.     createSchemaTables(config);
  84.     modifySchema(config);
  85.   }
  86. }

2、调用API，对hbase表进行保存|更新|删除操作

  1. 1、新增保存操作
  2.     HTable hTable = new HTable(configInit(), Bytes.toBytes(tableName));
  3.         //HTablePool hTablePool = new HTablePool(configInit(), 300);
  4.         try {
  5.             Put put = new Put(Bytes.toBytes(rowKey));
  6.             put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier), Bytes.toBytes(saveValue));
  7.             //hTablePool.getTable(tableName).put(put);
  8.             hTable.put(put);
  9.             return true;
  10.         } catch (Exception e) {
  11.             log.error("save error:{}, columnQualifier:{}, saveValue:{} ", e.getMessage(), columnFamily, saveValue);
  12.         } finally {
  13.             //hTablePool.closeTablePool(tableName);
  14.             Table.closeScanner(hTable, null);
  15.         }
  17. 2、更新操作
  18.     HTable hTable = new HTable(configInit(), TABLE_NAME);
  19.         try {
  20.             Put put = new Put(Bytes.toBytes(rowKey));
  21.             put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier), Bytes.toBytes(updateValue));
  22.             hTable.put(put);
  23.             return true;
  24.         } catch (Exception e) {
  25.             log.error("delete " + tableName + " error. save=" + columnQualifier, e);
  26.         } finally {
  27.             Table.closeScanner(hTable, null);
  28.         }
  30. 3、删除操作
  31.     HTable hTable = new HTable(configInit(), tableName);
  32.         try {
  33.             hTable.delete(delete);
  34.             return true;
  35.         } catch (Exception e) {
  36.             log.error("delete " + tableName + " error. delete=" + delete, e);
  37.         } finally {
  38.             Table.closeScanner(hTable, null);
  39.         }

3、使用hbaseTemplate, 对hbase表进行新增保存|更新|删除|查询操作

  1. 1、新增保存操作
  2.     hbaseTemplate.execute(tableName, table -> {
  3.                 boolean flag = false;
  4.                 try{
  5.                     byte[] rowkey = rowKey.getBytes();
  6.                     Put put = new Put(rowkey);
  7.                     put.add(Bytes.toBytes(columnFamily),Bytes.toBytes(tagCode), Bytes.toBytes(tagValue));
  8.                     table.put(put);
  9.                     flag = true;
  10.                 }catch(Exception e){
  11.                     log.error("保存hbase表数据失败,异常：{}, 列名：{}, 列值：{}", e.getMessage(), tagCode, tagValue);
  12.                 }
  13.                 return flag;
  14.     });
  16. 2、更新操作：跟新增保存一样，会自动覆盖值
  17.     hbaseTemplate.execute(tableName, table -> {
  18.             boolean flag = false;
  19.             try{
  20.                 byte[] rowkey = rowKey.getBytes();
  21.                 Put put = new Put(rowkey);
  22.                 put.add(Bytes.toBytes(columnFamily),Bytes.toBytes(tagCode), Bytes.toBytes(tagValue));
  23.                 table.put(put);
  24.                 flag = true;
  25.             }catch(Exception e){
  26.                 log.error("更新hbase表数据失败,异常：{}, 列名：{}", e.getMessage(), tagCode);
  27.             }
  28.             return flag;
  29.         });
  31. 3、删除操作
  32.     hbaseTemplate.execute(tableName, table -> {
  33.             boolean flag = false;
  34.             try{
  35.                 byte[] rowkey = rowKey.getBytes();
  36.                 //指定删除的rowKey
  37.                 Delete delete = new Delete(rowkey);
  38.                 //指定删除的列和列族
  39.                 delete.deleteColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(tagCode));
  40.                 table.delete(delete);
  41.                 flag = true;
  42.             }catch(Exception e){
  43.                 log.error("删除hbase表数据失败,异常：{}, 列名：{}", e.getMessage(), tagCode);
  44.             }
  45.             return flag;
  46.         });
  48. 4、查询操作
  49.     4-1、点查查询
  50.         hbaseTemplate.execute(tableName, table -> {
  51.             Get get = new Get(Bytes.toBytes(rowName));
  52.             //根据列族和列(接口参数传入的列)获取指定列的数据
  53.             get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnParam));
  54.             Result result = null;
  55.             try {
  56.                 result = table.get(get);
  57.             } catch (IOException e) {
  58.                 e.getStackTrace();
  59.             }
  60.             List<KeyValue> keyValueList = result.list();
  61.             if (!CollectionUtils.isEmpty(keyValueList)) {
  62.                 for (KeyValue keyValue : keyValueList) {
  63.                     String keyColumn = Bytes.toString(keyValue.getQualifierArray(), keyValue.getQualifierOffset(), keyValue.getQualifierLength());
  64.                 }
  65.                 return dataMap;
  66.             } else {
  67.                 return Collections.EMPTY_MAP;
  68.             }
  69.         });
  71.     4-2、scan查询
  72.         Scan scan = new Scan();
  73.             //行键开始范围
  74.             scan.setStartRow(Bytes.toBytes(ufeScanStart));
  75.             //行键结束范围
  76.             scan.setStopRow(Bytes.toBytes(ufeScanEnd));
  77.             //最大结果数(客户端缓存的最大字节数)
  78.             scan.setMaxResultSize(ufeScanLimit);
  79.             //是否反向扫描(默认是正向扫描(false)[按照字典顺序从小到大的顺序读取的])
  80.             scan.setReversed(ufeScanReversed);
  81.             //......
  83.         hbaseTemplate.find(tableName, scan, results -> {
  84.             List<Map<String, Object>> resultList = new ArrayList<>();
  85.             Iterator<Result> iterator = results.iterator();
  86.             while (iterator.hasNext()) {
  87.                 Result result = iterator.next();
  88.                 List<KeyValue> keyValueScanList = result.list();
  89.                 //业务逻辑
  90.                 }
  91.             }
  92.             return resultList;
  93.         });