pom依赖

  1. <!-- hbase-client -->
  2. <dependency>
  3.     <groupId>org.apache.hbase</groupId>
  4.     <artifactId>hbase-client</artifactId>
  5.     <version>2.2.2</version>
  6. </dependency>

配置Hosts文件

  1. # 需要配置集群的ip和主机名映射
  2. # 注意：你的集群有多少台机器就要添加多少个
  3. 127.0.0.1 master
  4. 127.0.0.2 datanode1
  5. 127.0.0.3 datanode2

工具类

HBaseConn.java - 连接工具类

  1. package com.example.hbase.hbasedemo;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.hbase.HBaseConfiguration;
  5. import org.apache.hadoop.hbase.TableName;
  6. import org.apache.hadoop.hbase.client.Connection;
  7. import org.apache.hadoop.hbase.client.ConnectionFactory;
  8. import org.apache.hadoop.hbase.client.Table;
  10. public class HBaseConn {
  12.     private static final HBaseConn INSTANCE = new HBaseConn();
  13.     private static Configuration configuration;
  14.     private static Connection connection;
  17.     //无参构造
  18.     private HBaseConn(){
  19.         try{
  20.             if(configuration == null){
  21.                 //创建配置文件对象
  22.                 configuration = HBaseConfiguration.create();
  23.                 //加载zookeeper配置
  24.                 configuration.set("hbase.zookeeper.quorum", "127.0.0.1,127.0.0.2,127.0.0.3");
  25.                 configuration.set("hbase.zookeeper.property.clientPort", "2181");
  26.             }
  27.         }catch (Exception e){
  28.             e.printStackTrace();
  29.         }
  30.     }
  32.     /**
  33.      * 获取连接
  34.      * @return
  35.      */
  36.     private Connection getConnection(){
  37.         if(connection == null || connection.isClosed()){
  38.             try {
  39.                 connection = ConnectionFactory.createConnection(configuration);
  40.             }catch (Exception e){
  41.                 e.printStackTrace();
  42.             }
  43.         }
  44.         return connection;
  45.     }
  47.     /**
  48.      * 获取连接 - 静态方法
  49.      * @return
  50.      */
  51.     public static Connection getHBaseConnection(){
  52.         return INSTANCE.getConnection();
  53.     }
  55.     /**
  56.      * 获取表
  57.      * @param tableName
  58.      * @return
  59.      * @throws Exception
  60.      */
  61.     public static Table getTable(String tableName) throws Exception{
  62.         return INSTANCE.getConnection().getTable(TableName.valueOf(tableName));
  63.     }
  65.     /**
  66.      * 关闭连接
  67.      */
  68.     public static void closeConnection(){
  69.         if(connection != null){
  70.             try {
  71.                 connection.close();
  72.             }catch (Exception e){
  73.                 e.printStackTrace();
  74.             }
  75.         }
  76.     }
  78. }

HBaseUtil.java - 操作hbase工具类

  1. package com.example.hbase.hbasedemo;
  3. import org.apache.hadoop.hbase.*;
  4. import org.apache.hadoop.hbase.client.*;
  5. import org.apache.hadoop.hbase.filter.FilterList;
  6. import org.apache.hadoop.hbase.util.Bytes;
  7. import java.util.Arrays;
  8. import java.util.List;
  10. public class HBaseUtil {
  12.     /**
  13.      * 创建表
  14.      * @param tableName 表名
  15.      * @param cfs 列祖列族
  16.      * @return
  17.      */
  18.     public static boolean createTable(String tableName,String [] cfs){
  19.         try (HBaseAdmin admin = (HBaseAdmin)HBaseConn.getHBaseConnection().getAdmin()){
  20.             if (admin.tableExists(TableName.valueOf(tableName))) {
  21.                 return false;
  22.             }
  23.             //定义表描述对象
  24.             TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
  25.             //遍历列族数组
  26.             Arrays.stream(cfs).forEach(cf -> {
  27.                 //定义列族描述对象
  28.                 ColumnFamilyDescriptor columnFamily = ColumnFamilyDescriptorBuilder.of(cf);
  29.                 //给表添加列族信息
  30.                 tableDescriptorBuilder.setColumnFamily(columnFamily);
  31.             });
  32.             //创建表
  33.             admin.createTable(tableDescriptorBuilder.build());
  34.         }catch (Exception e){
  35.             e.printStackTrace();
  36.             return false;
  37.         }
  38.         return true;
  39.     }
  41.     /**
  42.      * 删除表
  43.      * @param tableName 表名
  44.      * @return
  45.      */
  46.     public static boolean deleteTable(String tableName){
  47.         try (HBaseAdmin admin = (HBaseAdmin)HBaseConn.getHBaseConnection().getAdmin()){
  48.             admin.disableTable(TableName.valueOf(tableName));
  49.             admin.deleteTable(TableName.valueOf(tableName));
  50.         }catch (Exception e){
  51.             e.printStackTrace();
  52.             return false;
  53.         }
  54.         return true;
  55.     }
  57.     /**
  58.      * 插入数据
  59.      * @param tableName 表名
  60.      * @param rowkey 唯一标识
  61.      * @param cfName 列族名
  62.      * @param qualifer 列标识
  63.      * @param data 数据
  64.      * @return
  65.      */
  66.     public static boolean putRow(String tableName,String rowkey,String cfName,String qualifer,String data){
  67.         try (Table table = HBaseConn.getTable(tableName)){
  68.             Put put = new Put(Bytes.toBytes(rowkey));
  69.             put.addColumn(Bytes.toBytes(cfName),Bytes.toBytes(qualifer),Bytes.toBytes(data));
  70.             table.put(put);
  71.         }catch (Exception e){
  72.             e.printStackTrace();
  73.             return false;
  74.         }
  75.         return true;
  76.     }
  78.     /**
  79.      * 批量插入
  80.      * @param tableName 表名
  81.      * @param puts 数据
  82.      * @return
  83.      */
  84.     public static boolean putRows(String tableName, List<Put> puts){
  85.         try (Table table = HBaseConn.getTable(tableName)){
  86.             table.put(puts);
  87.         }catch (Exception e){
  88.             e.printStackTrace();
  89.             return false;
  90.         }
  91.         return true;
  92.     }
  94.     /**
  95.      * 获取单条数据
  96.      * @param tableName 表名
  97.      * @param rowKey 唯一标识
  98.      * @return
  99.      */
  100.     public static Result getRow(String tableName, String rowKey){
  101.         try (Table table = HBaseConn.getTable(tableName)){
  102.             Get get = new Get(Bytes.toBytes(rowKey));
  103.             return table.get(get);
  104.         }catch (Exception e){
  105.             e.printStackTrace();
  106.         }
  107.         return null;
  108.     }
  110.     /**
  111.      * 使用过滤器
  112.      * @param tableName 表名
  113.      * @param rowKey 唯一标识
  114.      * @param filterList 过滤器集合
  115.      * @return
  116.      */
  117.     public static Result getRow(String tableName, String rowKey, FilterList filterList){
  118.         try (Table table = HBaseConn.getTable(tableName)){
  119.             Get get = new Get(Bytes.toBytes(rowKey));
  120.             get.setFilter(filterList);
  121.             return table.get(get);
  122.         }catch (Exception e){
  123.             e.printStackTrace();
  124.         }
  125.         return null;
  126.     }
  128.     /**
  129.      * 全表扫描
  130.      * @param tableName 表名
  131.      * @return
  132.      */
  133.     public static ResultScanner getScanner(String tableName){
  134.         try (Table table = HBaseConn.getTable(tableName)){
  135.             Scan scan = new Scan();
  136.             scan.setCaching(1000);
  137.             return table.getScanner(scan);
  138.         }catch (Exception e){
  139.             e.printStackTrace();
  140.         }
  141.         return null;
  142.     }
  144.     /**
  145.      * 批量检索数据 - 范围检索
  146.      * @param tableName 表名
  147.      * @param startRowKey 起始rowkey
  148.      * @param endRowKey 终止rowkey
  149.      * @return
  150.      */
  151.     public static ResultScanner getScanner(String tableName,String startRowKey,String endRowKey){
  152.         try (Table table = HBaseConn.getTable(tableName)){
  153.             Scan scan = new Scan();
  154.             scan.withStartRow(Bytes.toBytes(startRowKey));
  155.             scan.withStopRow(Bytes.toBytes(endRowKey));
  156.             scan.setCaching(1000);
  158.             //取出数据
  159. //            ResultScanner resultScanner = table.getScanner(scan);
  160. //            for (Result rs : resultScanner) {
  161. //                Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("列族"),Bytes.toBytes("列1"))));
  162. //                Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("列族"),Bytes.toBytes("列2"))));
  163. //            }
  165.             return table.getScanner(scan);
  166.         }catch (Exception e){
  167.             e.printStackTrace();
  168.         }
  169.         return null;
  170.     }
  172.     /**
  173.      * 批量检索数据 - 使用过滤器
  174.      * @param tableName 表名
  175.      * @param startRowKey 起始rowkey
  176.      * @param endRowKey 终止rowkey
  177.      * @return
  178.      */
  179.     public static ResultScanner getScanner(String tableName,String startRowKey,String endRowKey,FilterList filterList){
  180.         /**
  181.          * FilterList用法
  182.          * FilterList.Operator.MUST_PASS_ALL - 满足所有条件 - 相当于sql中的and
  183.          * FilterList.Operator.MUST_PASS_ONE - 满足一个条件 - 相当于sql中的or
  184.          */
  185. //        Scan scan = new Scan();
  186. //        //创建过滤器集合
  187. //        FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
  188. //        //条件过滤器
  189. //        //CompareOperator.EQUAL - 绝对匹配
  190. //        SingleColumnValueFilter filter = new SingleColumnValueFilter(
  191. //                Bytes.toBytes("列族"),
  192. //                Bytes.toBytes("列"),
  193. //                CompareOperator.EQUAL,
  194. //                Bytes.toBytes("条件")
  195. //        );
  196. //        //添加过滤器
  197. //        filters.addFilter(filter);
  198. //
  199. //        //前缀过滤器
  200. //        PrefixFilter filter2 = new PrefixFilter(Bytes.toBytes("rowkey的前缀，比如井筒"));
  201. //        filters.addFilter(filter2);
  203.         try (Table table = HBaseConn.getTable(tableName)){
  204.             Scan scan = new Scan();
  205.             scan.withStartRow(Bytes.toBytes(startRowKey));
  206.             scan.withStopRow(Bytes.toBytes(endRowKey));
  207.             scan.setFilter(filterList);
  208.             scan.setCaching(1000);
  209.             return table.getScanner(scan);
  210.         }catch (Exception e){
  211.             e.printStackTrace();
  212.         }
  213.         return null;
  214.     }
  216.     /**
  217.      * 删除一行记录
  218.      * @param tableName 表名
  219.      * @param rowKey 唯一标识
  220.      * @return
  221.      */
  222.     public static boolean deleteRow(String tableName,String rowKey){
  223.         try (Table table = HBaseConn.getTable(tableName)){
  224.             Delete delete = new Delete(Bytes.toBytes(rowKey));
  225.             table.delete(delete);
  226.         }catch (Exception e){
  227.             e.printStackTrace();
  228.             return false;
  229.         }
  230.         return true;
  231.     }
  233.     /**
  234.      * 删除列族
  235.      * @param tableName 表名
  236.      * @param cfName 列族
  237.      * @return
  238.      */
  239.     public static boolean deleteColumnFamily(String tableName,String cfName){
  240.         try (HBaseAdmin admin = (HBaseAdmin)HBaseConn.getHBaseConnection().getAdmin()){
  241.             admin.deleteColumn(TableName.valueOf(tableName),Bytes.toBytes(cfName));
  242.         }catch (Exception e){
  243.             e.printStackTrace();
  244.             return false;
  245.         }
  246.         return true;
  247.     }
  249.     /**
  250.      * 删除列族
  251.      * @param tableName 表名
  252.      * @param cfName 列族
  253.      * @return
  254.      */
  255.     public static boolean deleteQualifier(String tableName,String rowKey,String cfName,String qualifier){
  256.         try (Table table = HBaseConn.getTable(tableName)){
  257.             Delete delete = new Delete(Bytes.toBytes(rowKey));
  258.             delete.addColumn(Bytes.toBytes(cfName),Bytes.toBytes(qualifier));
  259.             table.delete(delete);
  260.         }catch (Exception e){
  261.             e.printStackTrace();
  262.             return false;
  263.         }
  264.         return true;
  265.     }
  266. }

工具类的使用

  1. package com.example.hbase.hbasedemo;
  3. import org.apache.hadoop.hbase.CompareOperator;
  4. import org.apache.hadoop.hbase.client.Connection;
  5. import org.apache.hadoop.hbase.client.Put;
  6. import org.apache.hadoop.hbase.client.Result;
  7. import org.apache.hadoop.hbase.filter.*;
  8. import org.apache.hadoop.hbase.util.Bytes;
  9. import java.text.SimpleDateFormat;
  10. import java.util.*;
  12. public class HBaseTest {
  14.     private static Connection conn;
  15.     private static String testTableName = "TableTest";
  16.     static {
  17.         conn = HBaseConn.getHBaseConnection();
  18.         System.out.println("HBase连接 -> "+conn);
  19.     }
  21.     /**
  22.      * 创建并插入数据测试
  23.      * 以70个测点为例
  24.      * 插入1小时3600条数据，耗时30秒
  25.      */
  26.     public static void createTableAndDataTest() {
  27.         String [] cf = new String[]{"data","info"};//列族 - data数据 - info其他信息
  28.         boolean table = HBaseUtil.createTable(testTableName, cf);//创建表
  29.         String createTable = table?"成功":"失败";
  30.         System.out.println("创建表 -> " + createTable + " -> 开始插入");
  32.         //生成put，批量插入
  33.         List<Put> putList = new ArrayList<>();
  34.         long start = 1577808000000l;
  35.         for (int i = 0;i < 1 * 60 * 60; i++){
  36.             start+=1000;
  37.             String rowkey = "wellboreIdTest-"+(Long.MAX_VALUE - start);
  38.             Put put = new Put(Bytes.toBytes(rowkey));
  39.             //生成list列 - 70个列为例子
  40.             put.addColumn(Bytes.toBytes("data"),Bytes.toBytes("time"),Bytes.toBytes(timeStamp2Date(start,null)));
  41.             put.addColumn(Bytes.toBytes("data"),Bytes.toBytes("createTime"),Bytes.toBytes(timeStamp2Date(start,null)));
  42.             for (int j = 1; j < 70; j++) {
  43.                 String key = "key"+j;
  44.                 String value = "4500.00";
  45.                 put.addColumn(Bytes.toBytes("data"),Bytes.toBytes(key),Bytes.toBytes(value));
  46.             }
  47.             putList.add(put);
  48.         }
  50.         long s = System.currentTimeMillis();
  52.         boolean putRows = HBaseUtil.putRows(testTableName, putList);
  54.         long e = System.currentTimeMillis();
  56.         System.out.println("数据插入 -> " + putRows + " -> 耗时 -> " + (e-s));
  58.         HBaseConn.closeConnection();
  60.     }
  62.     /**
  63.      * 时间戳转字符串 - yyyy-MM-dd HH:mm:ss
  64.      * @param seconds
  65.      * @param format
  66.      * @return
  67.      */
  68.     public static String timeStamp2Date(long seconds,String format) {
  69.         if(format == null || format.isEmpty()){
  70.             format = "yyyy-MM-dd HH:mm:ss";
  71.         }
  72.         SimpleDateFormat sdf = new SimpleDateFormat(format);
  73.         return sdf.format(new Date(seconds));
  74.     }
  77.     /**
  78.      * 删除表
  79.      */
  80.     public static boolean deleteTable() {
  81.         boolean b = HBaseUtil.deleteTable(testTableName);
  82.         System.out.println("删除表 -> " + b);
  83.         return b;
  84.     }
  86.     /**
  87.      * 根据rowkey获取单条数据
  88.      */
  89.     public static void getOne() {
  90.         String rowkey = "wellboreIdTest-"+(Long.MAX_VALUE - 1577808002000l);
  91.         //获取数据
  92.         Result result = HBaseUtil.getRow(testTableName, rowkey);
  94.         Map<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes("data"));
  95.         for(Map.Entry<byte[], byte[]> entry:familyMap.entrySet()){
  96.             System.out.println(Bytes.toString(entry.getKey()));
  97.             System.out.println(Bytes.toString(entry.getValue()));
  98.         }
  99.     }
  101.     /*----------------------------- 过滤器用法 --------------------------*/
  103.     /**
  104.      * RowFilter 过滤器
  105.      * 比较过滤器
  106.      * @throws Exception
  107.      */
  108.     public void rowFilter()throws Exception {
  109.         //RowFilter过滤器使用 - CompareOperator比较运算符
  110.         Filter filter = new RowFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("rowkey1")));
  111.         //Operator.MUST_PASS_ALL 集合中的filter必须全部生效通过，才算通过
  112.         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, Arrays.asList(filter));
  113.     }
  115.     /**
  116.      * prefixFilter 过滤器
  117.      * 以 rowkey2 为前缀的所有行
  118.      * @throws Exception
  119.      */
  120.     public void prefixFilter()throws Exception {
  121.         Filter filter = new PrefixFilter(Bytes.toBytes("rowkey2"));
  122.         //Operator.MUST_PASS_ALL 集合中的filter必须全部生效通过，才算通过
  123.         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, Arrays.asList(filter));
  124.     }
  126.     /**
  127.      * keyOnlyFilter 过滤器
  128.      * 只返回 rowkey 和列的值 ，不会返回数据
  129.      * @throws Exception
  130.      */
  131.     public void keyOnlyFilter()throws Exception {
  132.         Filter filter = new KeyOnlyFilter(true);
  133.         //Operator.MUST_PASS_ALL 集合中的filter必须全部生效通过，才算通过
  134.         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, Arrays.asList(filter));
  135.     }
  137.     /**
  138.      * columnPrefixFilter 过滤器
  139.      * 列标识前缀过滤
  140.      * @throws Exception
  141.      */
  142.     public void columnPrefixFilter()throws Exception {
  143.         Filter filter = new ColumnPrefixFilter(Bytes.toBytes("nam"));
  144.         //Operator.MUST_PASS_ALL 集合中的filter必须全部生效通过，才算通过
  145.         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, Arrays.asList(filter));
  146.     }
  148. }