方式1：SpringBoot + HBaseUtil

1. 添加pom.xml依赖

  1. <dependency>
  2.     <groupId>org.apache.hbase</groupId>
  3.     <artifactId>hbase-client</artifactId>
  4.     <version>1.2.5</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.hbase</groupId>
  8.     <artifactId>hbase-server</artifactId>
  9.     <version>1.2.5</version>
  10. </dependency>

2. 编写关键类

■ 常量类（Constant.java）

  1. public class Constant {
  3.     /** HBase配置 **/
  4.     public static class HBaseConfig {
  5.         public final static String ZK_HOST="192.168.0.101,192.168.0.102,192.168.0.103";
  6.         public final static String ZK_PORT="2181";
  7.         public final static String ZK_PATH="/hbase";
  8.     }
  10. }

■ 工具类（HBaseUtil.java）

  1. import com.lonton.bigdata.constant.Constant;
  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.util.Bytes;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8. import java.io.IOException;
  9. import java.util.ArrayList;
  10. import java.util.List;
  12. public class HBaseUtil {
  14.     private static final Logger log = LoggerFactory.getLogger(HBaseUtil.class);
  16.     public static Configuration conf;
  17.     private static Connection conn;
  19.     static {
  20.         // 使用HBaseConfiguration的单例方法实例化
  21.         conf = HBaseConfiguration.create();
  22.         // HBase使用ZooKeeper的地址
  23.         conf.set(HConstants.ZOOKEEPER_QUORUM, Constant.HBaseConfig.ZK_HOST);
  24.         // ZooKeeper客户端端口
  25.         conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, Constant.HBaseConfig.ZK_PORT);
  26.         // ZooKeeper存储HBase信息的路径
  27.         conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, Constant.HBaseConfig.ZK_PATH);
  28.         // 所有客户端的默认scan缓存时间， 默认100
  29.         conf.set(HConstants.HBASE_CLIENT_SCANNER_CACHING, "500");
  30.         // 失败重试时等待时间， 默认100
  31.         conf.set(HConstants.HBASE_CLIENT_PAUSE, "50");
  32.         // 一次RPC请求的超时时间， 默认60000
  33.         conf.set(HConstants.HBASE_RPC_TIMEOUT_KEY, "2000");
  34.         // 客户端发起一次数据操作直至得到响应之间总的超时时间， 默认Integer.MAX_VALUE（一次阻塞操作）
  35.         conf.set(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, "3000");
  36.         // 重试次数3次，默认31
  37.         conf.set(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "3");
  38.     }
  40.     /**
  41.      * 获得链接
  42.      * @return
  43.      */
  44.     public static synchronized Connection getConnection() {
  45.         try {
  46.             if (conn == null || conn.isClosed()) {
  47.                 conn = ConnectionFactory.createConnection(conf);
  48.             }
  49.         } catch (IOException e) {
  50.             log.error("HBase 建立链接失败 ", e);
  51.         }
  52.         return conn;
  53.     }
  55.     /**
  56.      * 关闭连接
  57.      * @throws IOException
  58.      */
  59.     public static void closeConnect(Connection conn) {
  60.         if (null != conn) {
  61.             try {
  62.                 conn.close();
  63.             } catch (Exception e) {
  64.                 log.error("closeConnect failure !", e);
  65.             }
  66.         }
  67.     }
  69.     /**
  70.      * 判断表是否存在（DDL）
  71.      * create 't_user', {NAME => 'info', VERSIONS => '3'},{NAME => 'data'}
  72.      * @param tableName 表名
  73.      * @return 是否创建成功
  74.      * @throws IOException
  75.      */
  76.     public static boolean isTableExist(String tableName) throws IOException {
  77.         boolean result = false;
  78.         Connection conn = getConnection();
  79.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  80.         try {
  81.             result = admin.tableExists(tableName);
  82.         } catch (Exception e) {
  83.             log.error(e.getMessage());
  84.         } finally {
  85.             admin.close();
  86.         }
  87.         return result;
  88.     }
  90.     /**
  91.      * 创建表（DDL）
  92.      * create 't1', 'cf1', 'cf2'
  93.      * @param tableName    表名
  94.      * @param columnFamily 列簇
  95.      * @throws IOException
  96.      */
  97.     public static void createTable(String tableName, String... columnFamily) throws IOException {
  98.         // 判断表是否存在
  99.         if (isTableExist(tableName)) {
  100.             log.error("Table:" + tableName + " already exists!");
  101.             return;
  102.         }
  104.         Connection conn = getConnection();
  105.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  106.         try {
  107.             // 创建表属性对象,表名需要转字节
  108.             HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
  109.             // 创建多个列族
  110.             for (String cf : columnFamily) {
  111.                 descriptor.addFamily(new HColumnDescriptor(cf));
  112.             }
  113.             // 根据对表的配置，创建表
  114.             admin.createTable(descriptor);
  115.             log.info("Table:" + tableName + " create successfully!");
  116.         } catch (Exception e) {
  117.             log.error(e.getMessage());
  118.         } finally {
  119.             admin.close();
  120.         }
  121.     }
  123.     /**
  124.      * 新增记录（DML）
  125.      * put 't1', '1001', 'cf1:name', 'zhangsan'
  126.      * put 't1', '1001', 'cf1:age', '23'
  127.      * @param tableName    表名
  128.      * @param rowKey       行键
  129.      * @param columnFamily 列簇
  130.      * @param column       列
  131.      * @param value        值
  132.      * @throws IOException
  133.      */
  134.     public static void addRowData(String tableName, String rowKey, String columnFamily, String
  135.             column, String value) throws IOException {
  136.         Connection conn = getConnection();
  137.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  138.         try {
  139.             // 向表中插入数据
  140.             Put put = new Put(Bytes.toBytes(rowKey));
  141.             // 向Put对象中组装数据
  142.             put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
  143.             hTable.put(put);
  144.             log.info("insert successfully!");
  145.         } catch (Exception e) {
  146.             log.error(e.getMessage());
  147.         } finally {
  148.             hTable.close();
  149.         }
  150.     }
  152.     /**
  153.      * 全表扫描（Scan）
  154.      * scan "t1"
  155.      * @param tableName 表名
  156.      * @return
  157.      * @throws IOException
  158.      */
  159.     public static ResultScanner getAllRows(String tableName) throws IOException {
  160.         ResultScanner results = null;
  161.         Connection conn = getConnection();
  162.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  163.         try {
  164.             // 得到用于扫描region的对象
  165.             Scan scan = new Scan();
  166.             // setCaching设置的值为每次rpc的请求记录数，默认是1；cache大可以优化性能，但是太大了会花费很长的时间进行一次传输。
  167.             scan.setCaching(1000);
  168.             // 使用HTable得到resultcanner实现类的对象
  169.             results = hTable.getScanner(scan);
  170.             /*for (Result result : results) {
  171.                 Cell[] cells = result.rawCells();
  172.                 for (Cell cell : cells) {
  173.                     System.out.print("RowKey:" + Bytes.toString(CellUtil.cloneRow(cell)) + "\t");
  174.                     System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  175.                     System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  176.                     System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
  177.                     System.out.println();
  178.                 }
  179.             }*/
  180.         } catch (Exception e) {
  181.             log.error(e.getMessage());
  182.         } finally {
  183.             hTable.close();
  184.         }
  185.         return results;
  186.     }
  188.     /**
  189.      * 获取单行记录（get）
  190.      * get "t1","1001"
  191.      * @param tableName 表名
  192.      * @param rowKey    行键
  193.      * @return
  194.      * @throws IOException
  195.      */
  196.     public static Result getRow(String tableName, String rowKey) throws IOException {
  197.         Result result = null;
  198.         Connection conn = getConnection();
  199.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  200.         try {
  201.             Get get = new Get(Bytes.toBytes(rowKey));
  202.             //get.setMaxVersions(); // 显示所有版本
  203.             //get.setTimeStamp(); // 显示指定时间戳的版本
  204.             result = hTable.get(get);
  205.             /*for (Cell cell : result.rawCells()) {
  206.                 System.out.print("RowKey:" + Bytes.toString(result.getRow()) + "\t");
  207.                 System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  208.                 System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  209.                 System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)) + "\t");
  210.                 System.out.print("Timestamp:" + cell.getTimestamp());
  211.                 System.out.println();
  212.             }*/
  213.         } catch (Exception e) {
  214.             log.error(e.getMessage());
  215.         } finally {
  216.             hTable.close();
  217.         }
  218.         return result;
  219.     }
  221.     /**
  222.      * 获取多行记录（get）
  223.      * @param tableName 表名
  224.      * @param rows      行键（可扩展参数）
  225.      * @param <T>
  226.      * @return
  227.      * @throws IOException
  228.      */
  229.     public static <T> Result[] getRows(String tableName, List<T> rows) throws IOException {
  230.         Connection conn = getConnection();
  231.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  232.         List<Get> gets = null;
  233.         Result[] results = null;
  234.         try {
  235.             if (hTable != null) {
  236.                 gets = new ArrayList<Get>();
  237.                 for (T row : rows) {
  238.                     if (row != null) {
  239.                         gets.add(new Get(Bytes.toBytes(String.valueOf(row))));
  240.                     } else {
  241.                         throw new RuntimeException("hbase have no data");
  242.                     }
  243.                 }
  244.             }
  245.             if (gets.size() > 0) {
  246.                 results = hTable.get(gets);
  247.             }
  248.         } catch (IOException e) {
  249.             log.error("getRows failure !", e);
  250.         } finally {
  251.             try {
  252.                 hTable.close();
  253.             } catch (IOException e) {
  254.                 log.error("table.close() failure !", e);
  255.             }
  256.         }
  257.         return results;
  258.     }
  260.     /**
  261.      * 根据限定符获取单行记录（get）
  262.      * get "t1","1001","cf1:name"
  263.      * @param tableName 表名
  264.      * @param rowKey    行键
  265.      * @param family    列簇
  266.      * @param qualifier 限定符
  267.      * @throws IOException
  268.      */
  269.     public static Result getRowQualifier(String tableName, String rowKey, String family, String qualifier) throws IOException {
  270.         Result result = null;
  271.         Connection conn = getConnection();
  272.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  273.         try {
  274.             Get get = new Get(Bytes.toBytes(rowKey));
  275.             get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
  276.             result = hTable.get(get);
  277.             /*for (Cell cell : result.rawCells()) {
  278.                 System.out.print("RowKey:" + Bytes.toString(result.getRow()) + "\t");
  279.                 System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  280.                 System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  281.                 System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)) + "\t");
  282.                 System.out.print("Timestamp:" + cell.getTimestamp());
  283.                 System.out.println();
  284.             }*/
  285.         } catch (Exception e) {
  286.             log.error(e.getMessage());
  287.         } finally {
  288.             hTable.close();
  289.         }
  290.         return result;
  291.     }
  293.     /**
  294.      * 删除多行数据（DML）
  295.      * @param tableName 表名
  296.      * @param rows      行键（可扩展参数）
  297.      * @throws IOException
  298.      */
  299.     public static void deleteMultiRow(String tableName, String... rows) throws IOException {
  300.         Connection conn = getConnection();
  301.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  302.         try {
  303.             List<Delete> deleteList = new ArrayList<>();
  304.             for (String row : rows) {
  305.                 Delete delete = new Delete(Bytes.toBytes(row));
  306.                 deleteList.add(delete);
  307.             }
  308.             hTable.delete(deleteList);
  309.         } catch (Exception e) {
  310.             log.error(e.getMessage());
  311.         } finally {
  312.             hTable.close();
  313.         }
  314.     }
  316.     /**
  317.      * 删除表（DDL）
  318.      * @param tableName 表名
  319.      * @throws IOException
  320.      */
  321.     public static void dropTable(String tableName) throws IOException {
  322.         if (!isTableExist(tableName)) {
  323.             log.error("Table:" + tableName + " not exist!");
  324.             return;
  325.         }
  326.         Connection conn = getConnection();
  327.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  328.         try {
  329.             if (isTableExist(tableName)) {
  330.                 admin.disableTable(tableName);
  331.                 admin.deleteTable(tableName);
  332.                 log.info("Table:" + tableName + " delete successfully!");
  333.             }
  334.         } catch (Exception e) {
  335.             log.error(e.getMessage());
  336.         } finally {
  337.             admin.close();
  338.         }
  339.     }
  341. }

3. 测试

■ 测试类（HBaseBaseUtilTests.java）

  1. import org.apache.hadoop.hbase.Cell;
  2. import org.apache.hadoop.hbase.CellUtil;
  3. import org.apache.hadoop.hbase.client.Result;
  4. import org.apache.hadoop.hbase.client.ResultScanner;
  5. import org.apache.hadoop.hbase.util.Bytes;
  6. import org.junit.*;
  7. import java.io.IOException;
  8. import java.util.ArrayList;
  9. import java.util.List;
  11. public class HBaseBaseUtilTests {
  13.     private String tableName="t1";
  14.     private String rowKey="1001";
  16.     /** BeforeClass：会在所有方法被调用前被执行，
  17.      * 而且该方法是静态的，所有当测试类被加载后接着就会运行它，
  18.      * 而且在内存中它只会存在一份实例，它比较适合加载配置文件
  19.      **/
  20.     @BeforeClass
  21.     public static void setUpBeforeClass() {
  22.         System.out.println("this is @BeforeClass ...");
  23.         HBaseUtil.getConnection();
  24.     }
  26.     /** AfterClass：通常用来对资源的清理，如关闭数据库的连接 **/
  27.     @AfterClass
  28.     public static void tearDownAfterClass() {
  29.         System.out.println("this is @AfterClass ...");
  30.         HBaseUtil.closeConnect(HBaseUtil.getConnection());
  31.     }
  33.     /** Before：每个测试方法调用前执行一次 **/
  34.     @Before
  35.     public void setUp() {
  36.         System.out.println("this is @Before ...");
  37.     }
  39.     /** Before：每个测试方法调用后执行一次 **/
  40.     @After
  41.     public void tearDown() {
  42.         System.out.println("this is @After ...");
  43.     }
  45.     @Test
  46.     public void createTable() throws IOException {
  47.         HBaseUtil.createTable(tableName, "cf1", "cf2");
  48.         assert HBaseUtil.isTableExist(tableName);
  49.     }
  51.     @Test
  52.     public void isTableExist() throws IOException {
  53.         assert HBaseUtil.isTableExist(tableName);
  54.     }
  56.     @Test
  57.     public void addRowData() throws IOException {
  58.         String value="tom";
  59.         HBaseUtil.addRowData(tableName, rowKey, "cf1", "name", value);
  60.         Result result=HBaseUtil.getRowQualifier(tableName, rowKey, "cf1", "name");
  61.         Cell[] cells=result.rawCells();
  62.         Assert.assertNotNull(cells);
  63.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  64.     }
  66.     @Test
  67.     public void getAllRows() throws IOException {
  68.         ResultScanner results =HBaseUtil.getAllRows(tableName);
  69.         Assert.assertNotNull(results);
  70.         for (Result result : results) {
  71.             Assert.assertNotNull(result);
  72.         }
  73.     }
  75.     @Test
  76.     public void getRow() throws IOException {
  77.         String value="tom";
  78.         HBaseUtil.addRowData(tableName, rowKey, "cf1", "name", value);
  79.         Result result=HBaseUtil.getRow(tableName,rowKey);
  80.         Cell[] cells=result.rawCells();
  81.         Assert.assertNotNull(cells);
  82.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  83.     }
  85.     @Test
  86.     public void getRows() throws IOException {
  87.         String rowKey1="1003";
  88.         String rowKey2="1004";
  89.         String value1="tom1";
  90.         String value2="tom2";
  91.         HBaseUtil.addRowData(tableName, rowKey1, "cf1", "name", value1);
  92.         HBaseUtil.addRowData(tableName, rowKey2, "cf1", "name", value2);
  94.         List<String> rows=new ArrayList<>();
  95.         rows.add(rowKey1);
  96.         rows.add(rowKey2);
  98.         Result[] results=HBaseUtil.getRows(tableName,rows);
  99.         Assert.assertNotNull(results);
  100.         Assert.assertTrue(results.length==2);
  102.     }
  104.     @Test
  105.     public void getRowQualifier() throws IOException {
  106.         String value="tom";
  107.         HBaseUtil.addRowData(tableName, rowKey, "cf1", "name", value);
  108.         Result result=HBaseUtil.getRowQualifier(tableName, rowKey, "cf1", "name");
  109.         Cell[] cells=result.rawCells();
  110.         Assert.assertNotNull(cells);
  111.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  112.     }
  114.     @Test
  115.     public void deleteMultiRow() throws IOException {
  116.         String rowKey1="1003";
  117.         String rowKey2="1004";
  118.         String value1="tom1";
  119.         String value2="tom2";
  120.         HBaseUtil.addRowData(tableName, rowKey1, "cf1", "name", value1);
  121.         HBaseUtil.addRowData(tableName, rowKey2, "cf1", "name", value2);
  123.         HBaseUtil.deleteMultiRow(tableName, rowKey1, rowKey2);
  125.         List<String> rows=new ArrayList<>();
  126.         rows.add(rowKey1);
  127.         rows.add(rowKey2);
  129.         Result[] results=HBaseUtil.getRows(tableName,rows);
  130.         Assert.assertNotNull(results);
  131.         for (Result result : results) {
  132.             Cell[] cells=result.rawCells();
  133.             Assert.assertTrue(cells.length==0);
  134.         }
  135.     }
  137.     @Test
  138.     public void dropTable() throws IOException {
  139.         assert HBaseUtil.isTableExist(tableName);
  140.         HBaseUtil.dropTable(tableName);
  141.         assert !HBaseUtil.isTableExist(tableName);
  142.     }
  144. }

4. 完整示例代码

源码：hbase-springboot-example01-src.zip

方式2：SpringBoot + HBaseService

1. 添加pom.xml依赖

  1. <dependency>
  2.     <groupId>org.apache.hbase</groupId>
  3.     <artifactId>hbase-client</artifactId>
  4.     <version>1.2.5</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.hbase</groupId>
  8.     <artifactId>hbase-server</artifactId>
  9.     <version>1.2.5</version>
  10. </dependency>

2. 编写关键类

■ 常量类（Constant.java）

  1. public class Constant {
  3.     /** HBase配置 **/
  4.     public static class HBaseConfig {
  5.         public final static String ZK_HOST="192.168.0.101,192.168.0.102,192.168.0.103";
  6.         public final static String ZK_PORT="2181";
  7.         public final static String ZK_PATH="/hbase";
  8.     }
  10. }

■ 配置类（HBaseConfig.java）

  1. import com.lonton.bigdata.constant.Constant;
  2. import com.lonton.bigdata.service.HBaseService;
  3. import org.apache.hadoop.hbase.HBaseConfiguration;
  4. import org.apache.hadoop.hbase.HConstants;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. @Configuration
  9. public class HBaseConfig {
  11.     @Bean
  12.     public HBaseService getHBaseService() {
  13.         // 设置临时的hadoop环境变量，之后程序会去这个目录下的\bin目录下找winutils.exe工具，windows连接hadoop时会用到
  14.         //System.setProperty("hadoop.home.dir", "E:\\bigdataenv\\hadoop-common-2.2.0-bin-master");
  15.         // 执行此步时，会去resources目录下找相应的配置文件，例如hbase-site.xml
  16.         org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
  17.         // 使用HBaseConfiguration的单例方法实例化
  18.         // HBase使用ZooKeeper的地址
  19.         conf.set(HConstants.ZOOKEEPER_QUORUM, Constant.HBaseConfig.ZK_HOST);
  20.         // ZooKeeper客户端端口
  21.         conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, Constant.HBaseConfig.ZK_PORT);
  22.         // ZooKeeper存储HBase信息的路径
  23.         conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, Constant.HBaseConfig.ZK_PATH);
  24.         // 所有客户端的默认scan缓存时间， 默认100
  25.         conf.set(HConstants.HBASE_CLIENT_SCANNER_CACHING, "500");
  26.         // 失败重试时等待时间， 默认100
  27.         conf.set(HConstants.HBASE_CLIENT_PAUSE, "50");
  28.         // 一次RPC请求的超时时间， 默认60000
  29.         conf.set(HConstants.HBASE_RPC_TIMEOUT_KEY, "2000");
  30.         // 客户端发起一次数据操作直至得到响应之间总的超时时间， 默认Integer.MAX_VALUE（一次阻塞操作）
  31.         conf.set(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, "3000");
  32.         // 重试次数3次，默认31
  33.         conf.set(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "3");
  34.         return new HBaseService(conf);
  35.     }
  37. }

■ 服务类（HBaseService.java）

  1. import org.apache.hadoop.conf.Configuration;
  2. import org.apache.hadoop.hbase.*;
  3. import org.apache.hadoop.hbase.client.*;
  4. import org.apache.hadoop.hbase.util.Bytes;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import java.io.IOException;
  8. import java.util.ArrayList;
  9. import java.util.List;
  11. public class HBaseService {
  13.     private static final Logger log = LoggerFactory.getLogger(HBaseService.class);
  15.     public static Configuration conf;
  16.     private static Connection conn;
  18.     public HBaseService(Configuration conf){
  19.         try {
  20.             if (conn == null || conn.isClosed()) {
  21.                 conn = ConnectionFactory.createConnection(conf);
  22.             }
  23.         } catch (IOException e) {
  24.             log.error("HBase connect failure", e);
  25.         }
  26.     }
  28.     /**
  29.      * 获得链接
  30.      *
  31.      * @return
  32.      */
  33.     public Connection getConnection() {
  34.         try {
  35.             if (conn == null || conn.isClosed()) {
  36.                 conn = ConnectionFactory.createConnection(conf);
  37.             }
  38.         } catch (IOException e) {
  39.             log.error("HBase connect failure", e);
  40.         }
  41.         return conn;
  42.     }
  44.     /**
  45.      * 关闭连接
  46.      *
  47.      * @throws IOException
  48.      */
  49.     public void closeConnect(Connection conn) {
  50.         if (null != conn) {
  51.             try {
  52.                 conn.close();
  53.             } catch (Exception e) {
  54.                 log.error("closeConnect failure !", e);
  55.             }
  56.         }
  57.     }
  59.     /**
  60.      * 判断表是否存在（DDL）
  61.      * create 't_user', {NAME => 'info', VERSIONS => '3'},{NAME => 'data'}
  62.      *
  63.      * @param tableName 表名
  64.      * @return 是否创建成功
  65.      * @throws IOException
  66.      */
  67.     public boolean isTableExist(String tableName) throws IOException {
  68.         boolean result = false;
  69.         Connection conn = getConnection();
  70.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  71.         try {
  72.             result = admin.tableExists(tableName);
  73.         } catch (Exception e) {
  74.             log.error(e.getMessage());
  75.         } finally {
  76.             admin.close();
  77.         }
  78.         return result;
  79.     }
  81.     /**
  82.      * 创建表（DDL）
  83.      * create 't1', 'cf1', 'cf2'
  84.      *
  85.      * @param tableName    表名
  86.      * @param columnFamily 列簇
  87.      * @throws IOException
  88.      */
  89.     public void createTable(String tableName, String... columnFamily) throws IOException {
  90.         // 判断表是否存在
  91.         if (isTableExist(tableName)) {
  92.             log.error("Table:" + tableName + " already exists!");
  93.             return;
  94.         }
  96.         Connection conn = getConnection();
  97.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  98.         try {
  99.             // 创建表属性对象,表名需要转字节
  100.             HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
  101.             // 创建多个列族
  102.             for (String cf : columnFamily) {
  103.                 descriptor.addFamily(new HColumnDescriptor(cf));
  104.             }
  105.             // 根据对表的配置，创建表
  106.             admin.createTable(descriptor);
  107.             log.info("Table:" + tableName + " create successfully!");
  108.         } catch (Exception e) {
  109.             log.error(e.getMessage());
  110.         } finally {
  111.             admin.close();
  112.         }
  113.     }
  115.     /**
  116.      * 新增记录（DML）
  117.      * put 't1', '1001', 'cf1:name', 'zhangsan'
  118.      * put 't1', '1001', 'cf1:age', '23'
  119.      *
  120.      * @param tableName    表名
  121.      * @param rowKey       行键
  122.      * @param columnFamily 列簇
  123.      * @param column       列
  124.      * @param value        值
  125.      * @throws IOException
  126.      */
  127.     public void addRowData(String tableName, String rowKey, String columnFamily, String
  128.             column, String value) throws IOException {
  129.         Connection conn = getConnection();
  130.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  131.         try {
  132.             // 向表中插入数据
  133.             Put put = new Put(Bytes.toBytes(rowKey));
  134.             // 向Put对象中组装数据
  135.             put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
  136.             hTable.put(put);
  137.             log.info("insert successfully!");
  138.         } catch (Exception e) {
  139.             log.error(e.getMessage());
  140.         } finally {
  141.             hTable.close();
  142.         }
  143.     }
  145.     /**
  146.      * 全表扫描（Scan）
  147.      * scan "t1"
  148.      *
  149.      * @param tableName 表名
  150.      * @return
  151.      * @throws IOException
  152.      */
  153.     public ResultScanner getAllRows(String tableName) throws IOException {
  154.         ResultScanner results = null;
  155.         Connection conn = getConnection();
  156.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  157.         try {
  158.             // 得到用于扫描region的对象
  159.             Scan scan = new Scan();
  160.             // setCaching设置的值为每次rpc的请求记录数，默认是1；cache大可以优化性能，但是太大了会花费很长的时间进行一次传输。
  161.             scan.setCaching(1000);
  162.             // 使用HTable得到resultcanner实现类的对象
  163.             results = hTable.getScanner(scan);
  164.             /*for (Result result : results) {
  165.                 Cell[] cells = result.rawCells();
  166.                 for (Cell cell : cells) {
  167.                     System.out.print("RowKey:" + Bytes.toString(CellUtil.cloneRow(cell)) + "\t");
  168.                     System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  169.                     System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  170.                     System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
  171.                     System.out.println();
  172.                 }
  173.             }*/
  174.         } catch (Exception e) {
  175.             log.error(e.getMessage());
  176.         } finally {
  177.             hTable.close();
  178.         }
  179.         return results;
  180.     }
  182.     /**
  183.      * 获取单行记录（get）
  184.      * get "t1","1001"
  185.      *
  186.      * @param tableName 表名
  187.      * @param rowKey    行键
  188.      * @return
  189.      * @throws IOException
  190.      */
  191.     public Result getRow(String tableName, String rowKey) throws IOException {
  192.         Result result = null;
  193.         Connection conn = getConnection();
  194.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  195.         try {
  196.             Get get = new Get(Bytes.toBytes(rowKey));
  197.             //get.setMaxVersions(); // 显示所有版本
  198.             //get.setTimeStamp(); // 显示指定时间戳的版本
  199.             result = hTable.get(get);
  200.             /*for (Cell cell : result.rawCells()) {
  201.                 System.out.print("RowKey:" + Bytes.toString(result.getRow()) + "\t");
  202.                 System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  203.                 System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  204.                 System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)) + "\t");
  205.                 System.out.print("Timestamp:" + cell.getTimestamp());
  206.                 System.out.println();
  207.             }*/
  208.         } catch (Exception e) {
  209.             log.error(e.getMessage());
  210.         } finally {
  211.             hTable.close();
  212.         }
  213.         return result;
  214.     }
  216.     /**
  217.      * 获取多行记录（get）
  218.      *
  219.      * @param tableName 表名
  220.      * @param rows      行键（可扩展参数）
  221.      * @param <T>
  222.      * @return
  223.      * @throws IOException
  224.      */
  225.     public <T> Result[] getRows(String tableName, List<T> rows) throws IOException {
  226.         Connection conn = getConnection();
  227.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  228.         List<Get> gets = null;
  229.         Result[] results = null;
  230.         try {
  231.             if (hTable != null) {
  232.                 gets = new ArrayList<Get>();
  233.                 for (T row : rows) {
  234.                     if (row != null) {
  235.                         gets.add(new Get(Bytes.toBytes(String.valueOf(row))));
  236.                     } else {
  237.                         throw new RuntimeException("hbase have no data");
  238.                     }
  239.                 }
  240.             }
  241.             if (gets.size() > 0) {
  242.                 results = hTable.get(gets);
  243.             }
  244.         } catch (IOException e) {
  245.             log.error("getRows failure !", e);
  246.         } finally {
  247.             try {
  248.                 hTable.close();
  249.             } catch (IOException e) {
  250.                 log.error("table.close() failure !", e);
  251.             }
  252.         }
  253.         return results;
  254.     }
  256.     /**
  257.      * 根据限定符获取单行记录（get）
  258.      * get "t1","1001","cf1:name"
  259.      *
  260.      * @param tableName 表名
  261.      * @param rowKey    行键
  262.      * @param family    列簇
  263.      * @param qualifier 限定符
  264.      * @throws IOException
  265.      */
  266.     public Result getRowQualifier(String tableName, String rowKey, String family, String qualifier) throws IOException {
  267.         Result result = null;
  268.         Connection conn = getConnection();
  269.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  270.         try {
  271.             Get get = new Get(Bytes.toBytes(rowKey));
  272.             get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
  273.             result = hTable.get(get);
  274.             /*for (Cell cell : result.rawCells()) {
  275.                 System.out.print("RowKey:" + Bytes.toString(result.getRow()) + "\t");
  276.                 System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  277.                 System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  278.                 System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)) + "\t");
  279.                 System.out.print("Timestamp:" + cell.getTimestamp());
  280.                 System.out.println();
  281.             }*/
  282.         } catch (Exception e) {
  283.             log.error(e.getMessage());
  284.         } finally {
  285.             hTable.close();
  286.         }
  287.         return result;
  288.     }
  290.     /**
  291.      * 删除多行数据（DML）
  292.      *
  293.      * @param tableName 表名
  294.      * @param rows      行键（可扩展参数）
  295.      * @throws IOException
  296.      */
  297.     public void deleteMultiRow(String tableName, String... rows) throws IOException {
  298.         Connection conn = getConnection();
  299.         HTable hTable = (HTable) conn.getTable(TableName.valueOf(tableName));
  300.         try {
  301.             List<Delete> deleteList = new ArrayList<>();
  302.             for (String row : rows) {
  303.                 Delete delete = new Delete(Bytes.toBytes(row));
  304.                 deleteList.add(delete);
  305.             }
  306.             hTable.delete(deleteList);
  307.         } catch (Exception e) {
  308.             log.error(e.getMessage());
  309.         } finally {
  310.             hTable.close();
  311.         }
  312.     }
  314.     /**
  315.      * 删除表（DDL）
  316.      *
  317.      * @param tableName 表名
  318.      * @throws IOException
  319.      */
  320.     public void dropTable(String tableName) throws IOException {
  321.         if (!isTableExist(tableName)) {
  322.             log.error("Table:" + tableName + " not exist!");
  323.             return;
  324.         }
  325.         Connection conn = getConnection();
  326.         HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
  327.         try {
  328.             if (isTableExist(tableName)) {
  329.                 admin.disableTable(tableName);
  330.                 admin.deleteTable(tableName);
  331.                 log.info("Table:" + tableName + " delete successfully!");
  332.             }
  333.         } catch (Exception e) {
  334.             log.error(e.getMessage());
  335.         } finally {
  336.             admin.close();
  337.         }
  338.     }
  340. }

3. 测试

■ 测试类（HBaseServiceApplicationTests.java）

  1. import com.lonton.bigdata.service.HBaseService;
  2. import org.apache.hadoop.hbase.Cell;
  3. import org.apache.hadoop.hbase.CellUtil;
  4. import org.apache.hadoop.hbase.client.Result;
  5. import org.apache.hadoop.hbase.client.ResultScanner;
  6. import org.apache.hadoop.hbase.util.Bytes;
  7. import org.junit.Assert;
  8. import org.junit.jupiter.api.Test;
  9. import org.junit.runner.RunWith;
  10. import org.springframework.boot.test.context.SpringBootTest;
  11. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  12. import javax.annotation.Resource;
  13. import java.io.IOException;
  14. import java.util.ArrayList;
  15. import java.util.List;
  17. @RunWith(SpringJUnit4ClassRunner.class)
  18. @SpringBootTest
  19. class HBaseServiceApplicationTests {
  21.     private String tableName="t1";
  22.     private String rowKey="1001";
  24.     @Resource
  25.     private HBaseService service;
  27.     @Test
  28.     public void isTableExist() throws IOException {
  29.         assert service.isTableExist(tableName);
  30.     }
  32.     @Test
  33.     public void addRowData() throws IOException {
  34.         String value="tom";
  35.         service.addRowData(tableName, rowKey, "cf1", "name", value);
  36.         Result result=service.getRowQualifier(tableName, rowKey, "cf1", "name");
  37.         Cell[] cells=result.rawCells();
  38.         Assert.assertNotNull(cells);
  39.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  40.     }
  42.     @Test
  43.     public void getAllRows() throws IOException {
  44.         ResultScanner results =service.getAllRows(tableName);
  45.         Assert.assertNotNull(results);
  46.         for (Result result : results) {
  47.             Cell[] cells = result.rawCells();
  48.             for (Cell cell : cells) {
  49.                 System.out.print("RowKey:" + Bytes.toString(CellUtil.cloneRow(cell)) + "\t");
  50.                 System.out.print("ColumnFamily:" + Bytes.toString(CellUtil.cloneFamily(cell)) + "\t");
  51.                 System.out.print("Qualifier:" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t");
  52.                 System.out.print("Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
  53.                 System.out.println();
  54.             }
  55.             Assert.assertNotNull(result);
  56.         }
  57.     }
  59.     @Test
  60.     public void getRow() throws IOException {
  61.         String value="tom";
  62.         service.addRowData(tableName, rowKey, "cf1", "name", value);
  63.         Result result=service.getRow(tableName,rowKey);
  64.         Cell[] cells=result.rawCells();
  65.         Assert.assertNotNull(cells);
  66.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  67.     }
  69.     @Test
  70.     public void getRows() throws IOException {
  71.         String rowKey1="1003";
  72.         String rowKey2="1004";
  73.         String value1="tom1";
  74.         String value2="tom2";
  75.         service.addRowData(tableName, rowKey1, "cf1", "name", value1);
  76.         service.addRowData(tableName, rowKey2, "cf1", "name", value2);
  78.         List<String> rows=new ArrayList<>();
  79.         rows.add(rowKey1);
  80.         rows.add(rowKey2);
  82.         Result[] results=service.getRows(tableName,rows);
  83.         Assert.assertNotNull(results);
  84.         Assert.assertTrue(results.length==2);
  86.     }
  88.     @Test
  89.     public void getRowQualifier() throws IOException {
  90.         String value="tom";
  91.         service.addRowData(tableName, rowKey, "cf1", "name", value);
  92.         Result result=service.getRowQualifier(tableName, rowKey, "cf1", "name");
  93.         Cell[] cells=result.rawCells();
  94.         Assert.assertNotNull(cells);
  95.         Assert.assertEquals(value,Bytes.toString(CellUtil.cloneValue(cells[0])));
  96.     }
  98.     @Test
  99.     public void deleteMultiRow() throws IOException {
  100.         String rowKey1="1003";
  101.         String rowKey2="1004";
  102.         String value1="tom1";
  103.         String value2="tom2";
  104.         service.addRowData(tableName, rowKey1, "cf1", "name", value1);
  105.         service.addRowData(tableName, rowKey2, "cf1", "name", value2);
  107.         service.deleteMultiRow(tableName, rowKey1, rowKey2);
  109.         List<String> rows=new ArrayList<>();
  110.         rows.add(rowKey1);
  111.         rows.add(rowKey2);
  113.         Result[] results=service.getRows(tableName,rows);
  114.         Assert.assertNotNull(results);
  115.         for (Result result : results) {
  116.             Cell[] cells=result.rawCells();
  117.             Assert.assertTrue(cells.length==0);
  118.         }
  119.     }
  121.     @Test
  122.     public void dropTable() throws IOException {
  123.         assert service.isTableExist(tableName);
  124.         service.dropTable(tableName);
  125.         assert !service.isTableExist(tableName);
  126.     }
  128. }

4. 完整示例代码

源码：hbase-springboot-example02-src.zip

方式3：SpringBoot + HbaseTemplate

1. 添加pom.xml依赖

  1. <dependency>
  2.     <groupId>org.springframework.data</groupId>
  3.     <artifactId>spring-data-hadoop</artifactId>
  4.     <version>2.5.0.RELEASE</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.hbase</groupId>
  8.     <artifactId>hbase-client</artifactId>
  9.     <version>1.2.5</version>
  10. </dependency>

2. 工程配置

■ SpringBoot配置（application.yml）

  1. hbase:
  2.   config:
  3.     hbase.zookeeper.quorum: 192.168.0.101,192.168.0.102,192.168.0.103
  4.     hbase.zookeeper.property.clientPort: 2181
  5.     zookeeper.znode.parent: /hbase
  6.     hbase.client.scanner.caching: 500
  7.     hbase.client.pause: 50
  8.     hbase.rpc.timeout: 2000
  9.     hbase.client.operation.timeout: 3000
  10.     hbase.client.retries.number: 3
  11.     hbase.rootdir: hdfs://bigdata-node1:9000/hbase

3. 编写关键类

■ 配置映射类（HBaseProperties.java）

  1. import org.springframework.boot.context.properties.ConfigurationProperties;
  2. import java.util.Map;
  4. @ConfigurationProperties(prefix = "hbase")
  5. public class HBaseProperties {
  7.     private Map<String, String> config;
  9.     public Map<String, String> getConfig() {
  10.         return config;
  11.     }
  13.     public void setConfig(Map<String, String> config) {
  14.         this.config = config;
  15.     }
  17. }

■ 配置类（HBaseConfig.java）

  1. import org.apache.hadoop.hbase.HBaseConfiguration;
  2. import org.springframework.boot.context.properties.EnableConfigurationProperties;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.data.hadoop.hbase.HbaseTemplate;
  6. import java.util.Map;
  7. import java.util.Set;
  9. @Configuration
  10. @EnableConfigurationProperties(HBaseProperties.class)
  11. public class HBaseConfig {
  13.     private final HBaseProperties properties;
  15.     public HBaseConfig(HBaseProperties properties) {
  16.         this.properties = properties;
  17.     }
  19.     @Bean
  20.     public HbaseTemplate hbaseTemplate() {
  21.         HbaseTemplate hbaseTemplate = new HbaseTemplate();
  22.         hbaseTemplate.setConfiguration(configuration());
  23.         hbaseTemplate.setAutoFlush(true);
  24.         return hbaseTemplate;
  25.     }
  27.     public org.apache.hadoop.conf.Configuration configuration() {
  28.         org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
  29.         Map<String, String> config = properties.getConfig();
  30.         Set<String> keySet = config.keySet();
  31.         for (String key : keySet) {
  32.             configuration.set(key, config.get(key));
  33.         }
  34.         return configuration;
  35.     }
  37. }

其他方式：

  1. @Configuration
  2. public class HBaseConfig {
  4.     @Value("${hbase.zookeeper.quorum}")
  5.     private String zookeeperQuorum;
  7.     @Value("${hbase.zookeeper.property.clientPort}")
  8.     private String clientPort;
  10.     @Value("${zookeeper.znode.parent}")
  11.     private String znodeParent;
  13.     @Bean
  14.     public HbaseTemplate hbaseTemplate() {
  15.         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
  16.         conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
  17.         conf.set("hbase.zookeeper.property.clientPort", clientPort);
  18.         conf.set("zookeeper.znode.parent", znodeParent);
  19.         return new HbaseTemplate(conf);
  20.     }
  22. }

■ 服务类（HBaseService.java）

  1. import org.apache.hadoop.hbase.*;
  2. import org.apache.hadoop.hbase.client.*;
  3. import org.apache.hadoop.hbase.util.Bytes;
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.data.hadoop.hbase.HbaseTemplate;
  8. import org.springframework.data.hadoop.hbase.RowMapper;
  9. import org.springframework.data.hadoop.hbase.TableCallback;
  10. import org.springframework.stereotype.Service;
  11. import java.io.IOException;
  12. import java.util.ArrayList;
  13. import java.util.HashMap;
  14. import java.util.List;
  15. import java.util.Map;
  17. @Service
  18. public class HBaseService {
  20.     private static final Logger log = LoggerFactory.getLogger(HBaseService.class);
  22.     @Autowired
  23.     private HbaseTemplate hbaseTemplate;
  25.     /**
  26.      * 判断表是否存在（DDL）
  27.      * create 't_user', {NAME => 'info', VERSIONS => '3'},{NAME => 'data'}
  28.      *
  29.      * @param tableName 表名
  30.      * @return 是否创建成功
  31.      * @throws IOException
  32.      */
  33.     public boolean isTableExist(String tableName) throws IOException {
  34.         boolean result = false;
  35.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  36.         try {
  37.             result = admin.tableExists(tableName);
  38.         } catch (Exception e) {
  39.             log.error(e.getMessage());
  40.         } finally {
  41.             admin.close();
  42.         }
  43.         return result;
  44.     }
  46.     /**
  47.      * 创建表（DDL）
  48.      * create 't1', 'cf1', 'cf2'
  49.      *
  50.      * @param tableName    表名
  51.      * @param columnFamily 列簇
  52.      * @throws IOException
  53.      */
  54.     public void createTable(String tableName, String... columnFamily) throws IOException {
  55.         // 判断表是否存在
  56.         if (isTableExist(tableName)) {
  57.             log.error("Table:" + tableName + " already exists!");
  58.             return;
  59.         }
  61.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  62.         try {
  63.             // 创建表属性对象,表名需要转字节
  64.             HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
  65.             // 创建多个列族
  66.             for (String cf : columnFamily) {
  67.                 descriptor.addFamily(new HColumnDescriptor(cf));
  68.             }
  69.             // 根据对表的配置，创建表
  70.             admin.createTable(descriptor);
  71.             log.info("Table:" + tableName + " create successfully!");
  72.         } catch (Exception e) {
  73.             log.error(e.getMessage());
  74.         } finally {
  75.             admin.close();
  76.         }
  77.     }
  79.     /**
  80.      * 新增记录（DML）
  81.      * put 't1', '1001', 'cf1:name', 'zhangsan'
  82.      * put 't1', '1001', 'cf1:age', '23'
  83.      *
  84.      * @param tableName    表名
  85.      * @param rowKey       行键
  86.      * @param columnFamily 列簇
  87.      * @param column       列
  88.      * @param value        值
  89.      */
  90.     public void addRowData(String tableName, String rowKey, String columnFamily, String
  91.             column, String value) {
  92.         hbaseTemplate.put(tableName, rowKey, columnFamily, column, Bytes.toBytes(value));
  93.     }
  95.     /**
  96.      * 全表扫描（Scan）
  97.      * scan 't1'
  98.      * [
  99.      * {cf2={name=zhangsan, age=23}, cf1={name=tom1, age=23}},
  100.      * {cf2={name=lisi, age=24}}
  101.      * ]
  102.      *
  103.      * @param tableName 表名
  104.      * @return
  105.      */
  106.     public List<Map<String, Map<String, String>>> getAllRows(String tableName) {
  107.         Scan scan = new Scan();
  108.         List<Map<String, Map<String, String>>> list = hbaseTemplate.find(tableName, scan, (result, rowNum) -> {
  109.             Cell[] cells = result.rawCells();
  110.             Map<String, Map<String, String>> data = new HashMap<>(16);
  111.             for (Cell c : cells) {
  112.                 String columnFamily = new String(CellUtil.cloneFamily(c));
  113.                 String rowName = new String(CellUtil.cloneQualifier(c));
  114.                 String value = new String(CellUtil.cloneValue(c));
  115.                 Map<String, String> obj = data.get(columnFamily);
  116.                 if (null == obj) {
  117.                     obj = new HashMap<>(16);
  118.                 }
  119.                 obj.put(rowName, value);
  120.                 data.put(columnFamily, obj);
  121.             }
  122.             return data;
  123.         });
  124.         return list;
  125.     }
  127.     /**
  128.      * 根据RowKey获取单行记录（get）
  129.      * get 't1','1001'
  130.      * {cf1_name=tom, cf1_age=23}
  131.      *
  132.      * @param tableName 表名
  133.      * @param rowKey    行键
  134.      * @return
  135.      */
  136.     public Map<String, Map<String, Object>> getRowByRowKey(String tableName, String rowKey) {
  137.         return hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String, Map<String, Object>>>() {
  138.             @Override
  139.             public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  140.                 Cell[] cells = result.rawCells();
  141.                 Map<String, Map<String, Object>> data = new HashMap<>(16);
  142.                 for (Cell c : cells) {
  143.                     String columnFamily = new String(CellUtil.cloneFamily(c));
  144.                     String rowName = new String(CellUtil.cloneQualifier(c));
  145.                     String value = new String(CellUtil.cloneValue(c));
  146.                     Map<String, Object> obj = data.get(columnFamily);
  147.                     if (null == obj) {
  148.                         obj = new HashMap<>(16);
  149.                     }
  150.                     obj.put(rowName, value);
  151.                     data.put(columnFamily, obj);
  152.                 }
  153.                 return data;
  154.             }
  155.         });
  156.     }
  158.     /**
  159.      * 获取多行记录（get）
  160.      *
  161.      * @param tableName 表名
  162.      * @param rowKeys   行键（可扩展参数）
  163.      * @return
  164.      */
  165.     public List<Map<String, Map<String, Object>>> getRows(String tableName, String... rowKeys) {
  166.         List<Map<String, Map<String, Object>>> rows = new ArrayList<>();
  167.         for (String rowKey : rowKeys) {
  168.             Map<String, Map<String, Object>> row = hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String, Map<String, Object>>>() {
  169.                 @Override
  170.                 public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  171.                     Cell[] cells = result.rawCells();
  172.                     Map<String, Map<String, Object>> data = new HashMap<>(16);
  173.                     for (Cell c : cells) {
  174.                         String columnFamily = new String(CellUtil.cloneFamily(c));
  175.                         String rowName = new String(CellUtil.cloneQualifier(c));
  176.                         String value = new String(CellUtil.cloneValue(c));
  177.                         Map<String, Object> obj = data.get(columnFamily);
  178.                         if (null == obj) {
  179.                             obj = new HashMap<>(16);
  180.                         }
  181.                         obj.put(rowName, value);
  182.                         obj.put("rowKey", rowKey);
  183.                         data.put(columnFamily, obj);
  184.                     }
  185.                     return data;
  186.                 }
  187.             });
  188.             rows.add(row);
  189.         }
  190.         return rows;
  191.     }
  193.     /**
  194.      * 根据限定符获取指定数据（get）
  195.      *
  196.      * @param tableName  表名
  197.      * @param rowName    行键
  198.      * @param familyName 列簇
  199.      * @param qualifier  限定符
  200.      * @return
  201.      */
  202.     public String getRowQualifier(String tableName, String rowName, String familyName, String qualifier) {
  203.         return hbaseTemplate.get(tableName, rowName, familyName, qualifier, (result, i) -> {
  204.             List<Cell> ceList = result.listCells();
  205.             String res = "";
  206.             if (ceList != null && ceList.size() > 0) {
  207.                 for (Cell cell : ceList) {
  208.                     res = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  209.                 }
  210.             }
  211.             return res;
  212.         });
  213.     }
  215.     /**
  216.      * 根据StartKey和EndKey获取数据（scan），前闭后开区间
  217.      *
  218.      * @param tableName 表名
  219.      * @param startRow  起始rowKey
  220.      * @param stopRow   结束rowKey
  221.      * @return
  222.      */
  223.     public List<Map<String, Map<String, Object>>> getRowsByKey(String tableName, String startRow, String stopRow) {
  224.         Scan scan = new Scan();
  225.         if (startRow == null) {
  226.             startRow = "";
  227.         }
  228.         if (stopRow == null) {
  229.             stopRow = "";
  230.         } else {
  231.             stopRow += "|";
  232.         }
  233.         scan.setStartRow(Bytes.toBytes(startRow));
  234.         scan.setStopRow(Bytes.toBytes(stopRow));
  235.         /* PageFilter filter = new PageFilter(5);
  236.          scan.setFilter(filter);*/
  237.         return hbaseTemplate.find(tableName, scan, new RowMapper<Map<String, Map<String, Object>>>() {
  238.             @Override
  239.             public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  240.                 List<Cell> ceList = result.listCells();
  241.                 Map<String, Map<String, Object>> data = new HashMap<>(16);
  242.                 if (ceList != null && ceList.size() > 0) {
  243.                     for (Cell cell : ceList) {
  244.                         String rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
  245.                         String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  246.                         String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  247.                         String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  248.                         Map<String, Object> obj = data.get(family);
  249.                         if (null == obj) {
  250.                             obj = new HashMap<>(16);
  251.                         }
  252.                         obj.put("rowKey", rowKey);
  253.                         obj.put(qualifier, value);
  254.                         data.put(family, obj);
  255.                     }
  256.                 }
  257.                 return data;
  258.             }
  259.         });
  260.     }
  262.     /**
  263.      * 删除列簇（DML）
  264.      *
  265.      * @param tableName  表名
  266.      * @param rowKey     行键
  267.      * @param familyName 列簇
  268.      */
  269.     public void deleteRow(String tableName, String rowKey, String familyName) {
  270.         hbaseTemplate.delete(tableName, rowKey, familyName);
  271.     }
  273.     /**
  274.      * 删除限定符（DML）
  275.      *
  276.      * @param tableName  表名
  277.      * @param rowKey     行键
  278.      * @param familyName 列簇
  279.      * @param qualifier  限定符
  280.      */
  281.     public void deleteQualifier(String tableName, String rowKey, String familyName, String qualifier) {
  282.         hbaseTemplate.delete(tableName, rowKey, familyName, qualifier);
  283.     }
  285.     /**
  286.      * 删除多行数据（DML）
  287.      *
  288.      * @param tableName   表名
  289.      * @param familyNames 列簇
  290.      * @param rowKeys     行键（可扩展参数）
  291.      */
  292.     public void deleteRows(String tableName, String[] familyNames, String... rowKeys) {
  293.         for (String rowKey : rowKeys) {
  294.             for (String familyName : familyNames) {
  295.                 hbaseTemplate.delete(tableName, rowKey, familyName);
  296.             }
  297.         }
  298.     }
  300.     /**
  301.      * 根据rowKey删除记录（DML）
  302.      *
  303.      * @param tableName 表名
  304.      * @param rowKey    行键
  305.      */
  306.     public void deleteAll(String tableName, final String rowKey) {
  307.         hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {
  308.             public Boolean doInTable(HTableInterface table) {
  309.                 boolean flag = false;
  310.                 try {
  311.                     List<Delete> list = new ArrayList<>();
  312.                     Delete d1 = new Delete(rowKey.getBytes());
  313.                     list.add(d1);
  314.                     table.delete(list);
  315.                     flag = true;
  316.                 } catch (Exception e) {
  317.                     e.printStackTrace();
  318.                 }
  319.                 return flag;
  320.             }
  321.         });
  322.     }
  324.     /**
  325.      * 删除表（DDL）
  326.      *
  327.      * @param tableName 表名
  328.      * @throws IOException
  329.      */
  330.     public void dropTable(String tableName) throws IOException {
  331.         if (!isTableExist(tableName)) {
  332.             log.error("Table:" + tableName + " not exist!");
  333.             return;
  334.         }
  335.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  336.         try {
  337.             if (isTableExist(tableName)) {
  338.                 admin.disableTable(tableName);
  339.                 admin.deleteTable(tableName);
  340.                 log.info("Table:" + tableName + " delete successfully!");
  341.             }
  342.         } catch (Exception e) {
  343.             log.error(e.getMessage());
  344.         } finally {
  345.             admin.close();
  346.         }
  347.     }
  349.     /**
  350.      * 通用方法（一般可用于更新操作，如：delete、put）
  351.      示例：deleteAll
  352.      TableCallback action = new TableCallback<Boolean>() {
  353.          public Boolean doInTable(HTableInterface table) {
  354.              boolean flag = false;
  355.              try {
  356.                  List<Delete> list = new ArrayList<>();
  357.                  Delete d1 = new Delete(rowKey.getBytes());
  358.                  list.add(d1);
  359.                  table.delete(list);
  360.                  flag = true;
  361.              } catch (Exception e) {
  362.                 e.printStackTrace();
  363.              }
  364.              return flag;
  365.          }
  366.      };
  367.      示例：put
  368.      TableCallback action = new TableCallback<Boolean>() {
  369.          public Boolean doInTable(HTableInterface table) {
  370.              boolean flag = false;
  371.              try {
  372.                  Put put = new Put(rowKey.getBytes());
  373.                  put.add(Bytes.toBytes(CF_01), Bytes.toBytes(CF1_QUALIFIER_NAME), Bytes.toBytes(CF1_NAME_VALUE));
  374.                  put.add(Bytes.toBytes(CF_02), Bytes.toBytes(CF2_QUALIFIER_ADDR), Bytes.toBytes(CF2_ADDR_VALUE));
  375.                  table.put(put);
  376.                  flag = true;
  377.              } catch (Exception e) {
  378.                 e.printStackTrace();
  379.              }
  380.              return flag;
  381.          }
  382.      };
  383.      * @param tableName
  384.      * @param action
  385.      */
  386.     public void execute(String tableName, TableCallback action) {
  387.         hbaseTemplate.execute(tableName, action);
  388.     }
  390. }

4. 测试

■ 测试类（HBaseServiceApplicationTests.java）

  1. import org.apache.hadoop.hbase.client.Delete;
  2. import org.apache.hadoop.hbase.client.HTableInterface;
  3. import org.apache.hadoop.hbase.client.Put;
  4. import org.apache.hadoop.hbase.util.Bytes;
  5. import org.junit.Assert;
  6. import org.junit.jupiter.api.Test;
  7. import org.junit.runner.RunWith;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.boot.test.context.SpringBootTest;
  10. import org.springframework.data.hadoop.hbase.TableCallback;
  11. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  12. import java.io.IOException;
  13. import java.util.ArrayList;
  14. import java.util.List;
  15. import java.util.Map;
  17. @RunWith(SpringJUnit4ClassRunner.class)
  18. @SpringBootTest
  19. class HBaseServiceApplicationTests {
  21.     private String tableName = "t1";
  22.     private String rowKey = "1001";
  23.     private String rowKey2 = "1002";
  24.     private String rowKey3 = "1003";
  26.     private String CF_01 = "cf1";
  27.     private String CF1_QUALIFIER_NAME = "name";
  28.     private String CF1_QUALIFIER_AGE = "age";
  29.     private String CF1_NAME_VALUE = "tom";
  30.     private String CF1_AGE_VALUE = "23";
  33.     private String CF_02 = "cf2";
  34.     private String CF2_QUALIFIER_ADDR = "address";
  35.     private String CF2_ADDR_VALUE = "beijing";
  37.     @Autowired
  38.     private HBaseService service;
  40.     @Test
  41.     public void isTableExist() throws IOException {
  42.         assert service.isTableExist(tableName);
  43.     }
  45.     @Test
  46.     public void createTable() throws IOException {
  47.         service.createTable(tableName, CF_01, CF_02);
  48.         assert service.isTableExist(tableName);
  49.     }
  51.     @Test
  52.     public void addRowData() {
  53.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  54.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  55.     }
  57.     @Test
  58.     public void getAllRows() {
  59.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  60.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  61.         List<Map<String, Map<String, String>>> rows = service.getAllRows(tableName);
  62.         Assert.assertTrue(rows.size() == 1); // 1个RowKey，对应1行数据
  63.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  64.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  65.         Assert.assertEquals(CF1_AGE_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_AGE));
  66.         System.out.println(rows);
  67.     }
  69.     @Test
  70.     public void getRowByRowKey() {
  71.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  72.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  73.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  74.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  75.         Assert.assertNotNull(row);
  76.         Assert.assertEquals(CF1_NAME_VALUE, row.get(CF_01).get(CF1_QUALIFIER_NAME));
  77.         Assert.assertEquals(CF1_AGE_VALUE, row.get(CF_01).get(CF1_QUALIFIER_AGE));
  78.         Assert.assertEquals(CF2_ADDR_VALUE, row.get(CF_02).get(CF2_QUALIFIER_ADDR));
  79.         System.out.println(row);
  80.     }
  82.     @Test
  83.     public void getRows() {
  84.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  85.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  86.         List<Map<String, Map<String, Object>>> rows = service.getRows(tableName, rowKey2, rowKey3);
  87.         Assert.assertTrue(rows.size() == 2); // 2个RowKey，对应1行数据
  88.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  89.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  90.         System.out.println(rows);
  91.     }
  93.     @Test
  94.     public void getRowQualifier() {
  95.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  96.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  97.     }
  99.     @Test
  100.     public void getRowsByKey() {
  101.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  102.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  103.         List<Map<String, Map<String, Object>>> rows = service.getRowsByKey(tableName, rowKey2, rowKey3);
  104.         Assert.assertTrue(rows.size() == 2); // 2个RowKey，对应1行数据
  105.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  106.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  107.         System.out.println(rows);
  108.     }
  110.     @Test
  111.     public void deleteQualifier() {
  112.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  113.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  114.         service.deleteQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME);
  115.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  116.         System.out.println(row);
  117.         Assert.assertNull(row.get(CF_01).get(CF1_QUALIFIER_NAME));
  118.         Assert.assertNotNull(row.get(CF_01).get(CF1_QUALIFIER_AGE));
  119.     }
  121.     @Test
  122.     public void deleteRow() {
  123.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  124.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  125.         service.deleteRow(tableName, rowKey, CF_01);
  126.         service.deleteRow(tableName, rowKey, CF_02);
  127.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  128.         System.out.println(row);
  129.         Assert.assertTrue(row.isEmpty());
  130.     }
  132.     @Test
  133.     public void deleteMultiRow() {
  134.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  135.         service.addRowData(tableName, rowKey2, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  136.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  137.         service.addRowData(tableName, rowKey3, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  138.         List<Map<String, Map<String, Object>>> rows = service.getRows(tableName, rowKey2, rowKey3);
  139.         System.out.println("----------------------- test 1 -------------------------");
  140.         System.out.println(rows);
  142.         String[] familyNames = {CF_01, CF_02};
  143.         service.deleteRows(tableName, familyNames, rowKey2, rowKey3);
  144.         rows = service.getRows(tableName, rowKey2, rowKey3);
  145.         System.out.println("----------------------- test 2 -------------------------");
  146.         System.out.println(rows); // [{}, {}]
  147.         Assert.assertTrue(rows.get(0).isEmpty());
  148.     }
  150.     @Test
  151.     public void deleteAll() {
  152.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  153.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  154.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  155.         System.out.println(row);
  156.         service.deleteAll(tableName, rowKey);
  157.         row = service.getRowByRowKey(tableName, rowKey);
  158.         System.out.println(row);
  159.         Assert.assertTrue(row.isEmpty());
  160.     }
  162.     @Test
  163.     public void dropTable() throws IOException {
  164.         if (!service.isTableExist(tableName)) {
  165.             service.createTable(tableName);
  166.         }
  167.         assert service.isTableExist(tableName);
  168.         service.dropTable(tableName);
  169.         assert !service.isTableExist(tableName);
  170.     }
  172.     @Test
  173.     public void execute_deleteAll() {
  174.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  175.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  176.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  177.         System.out.println(row);
  178.         TableCallback action = new TableCallback<Boolean>() {
  179.             public Boolean doInTable(HTableInterface table) {
  180.                 boolean flag = false;
  181.                 try {
  182.                     List<Delete> list = new ArrayList<>();
  183.                     Delete d1 = new Delete(rowKey.getBytes());
  184.                     list.add(d1);
  185.                     table.delete(list);
  186.                     flag = true;
  187.                 } catch (Exception e) {
  188.                     e.printStackTrace();
  189.                 }
  190.                 return flag;
  191.             }
  192.         };
  193.         service.execute(tableName, action);
  194.         row = service.getRowByRowKey(tableName, rowKey);
  195.         System.out.println(row);
  196.         Assert.assertTrue(row.isEmpty());
  197.     }
  199.     @Test
  200.     public void execute_put() {
  201.         TableCallback action = new TableCallback<Boolean>() {
  202.             public Boolean doInTable(HTableInterface table) {
  203.                 boolean flag = false;
  204.                 try {
  205.                     Put put = new Put(rowKey.getBytes());
  206.                     put.add(Bytes.toBytes(CF_01), Bytes.toBytes(CF1_QUALIFIER_NAME), Bytes.toBytes(CF1_NAME_VALUE));
  207.                     put.add(Bytes.toBytes(CF_02), Bytes.toBytes(CF2_QUALIFIER_ADDR), Bytes.toBytes(CF2_ADDR_VALUE));
  208.                     table.put(put);
  209.                     flag = true;
  210.                 } catch (Exception e) {
  211.                     e.printStackTrace();
  212.                 }
  213.                 return flag;
  214.             }
  215.         };
  216.         service.execute(tableName, action);
  217.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  218.         Assert.assertEquals(CF2_ADDR_VALUE, service.getRowQualifier(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR));
  219.     }
  221. }

5. 完整示例代码

源码：hbase-springboot-example03-src.zip

方式4：SpringBoot + HbaseTemplate（Spring配置）

1. 添加pom.xml依赖

  1. <dependency>
  2.     <groupId>org.springframework.data</groupId>
  3.     <artifactId>spring-data-hadoop</artifactId>
  4.     <version>2.5.0.RELEASE</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.hbase</groupId>
  8.     <artifactId>hbase-client</artifactId>
  9.     <version>1.2.5</version>
  10. </dependency>

2. 工程配置

■ 参数配置（resources/config/hbase.properties）

  1. hbase.zk.host=192.168.0.101,192.168.0.102,192.168.0.103
  2. hbase.zk.port=2181

■ Spring配置（resources/config/hbase-spring.xml）

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xmlns:hdp="http://www.springframework.org/schema/hadoop"
  6.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  7.     http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
  8.     http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
  10.     <context:property-placeholder location="classpath:/config/hbase.properties"/>
  12.     <hdp:configuration id="hadoopConfiguration">fs.default.name=hdfs://bigdata-node1:9000</hdp:configuration>
  13.     <!-- HA -->
  14.     <!--<hdp:configuration id="hadoopConfiguration">fs.defaultFS=hdfs://bigdata-node1:9000</hdp:configuration>-->
  16.     <hdp:hbase-configuration id="hbaseConfiguration" configuration-ref="hadoopConfiguration" zk-quorum="${hbase.zk.host}" zk-port="${hbase.zk.port}" delete-connection="false" />
  18.     <bean id="hbaseTemplate" class="org.springframework.data.hadoop.hbase.HbaseTemplate">
  19.         <property name="configuration" ref="hbaseConfiguration"/>
  20.     </bean>
  22. </beans>

3. 编写关键类

■ 启动类（Application.java）

  1. import org.springframework.boot.SpringApplication;
  2. import org.springframework.boot.autoconfigure.SpringBootApplication;
  3. import org.springframework.context.annotation.ImportResource;
  5. @SpringBootApplication
  6. @ImportResource(locations = {"classpath:/config/hbase-spring.xml"})
  7. public class Application {
  9.     public static void main(String[] args) {
  10.         // 设置临时的hadoop环境变量，之后程序会去这个目录下的\bin目录下找winutils.exe工具，windows连接hadoop时会用到
  11.         //System.setProperty("hadoop.home.dir", "E:\\bigdataenv\\hadoop-common-2.2.0-bin-master");
  12.         SpringApplication.run(Application.class, args);
  13.     }
  15. }

■ 服务类（HBaseService.java）

  1. import org.apache.hadoop.hbase.*;
  2. import org.apache.hadoop.hbase.client.*;
  3. import org.apache.hadoop.hbase.util.Bytes;
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.data.hadoop.hbase.HbaseTemplate;
  8. import org.springframework.data.hadoop.hbase.RowMapper;
  9. import org.springframework.data.hadoop.hbase.TableCallback;
  10. import org.springframework.stereotype.Service;
  11. import java.io.IOException;
  12. import java.util.ArrayList;
  13. import java.util.HashMap;
  14. import java.util.List;
  15. import java.util.Map;
  17. @Service
  18. public class HBaseService {
  20.     private static final Logger log = LoggerFactory.getLogger(HBaseService.class);
  22.     @Autowired
  23.     private HbaseTemplate hbaseTemplate;
  25.     /**
  26.      * 判断表是否存在（DDL）
  27.      * create 't_user', {NAME => 'info', VERSIONS => '3'},{NAME => 'data'}
  28.      *
  29.      * @param tableName 表名
  30.      * @return 是否创建成功
  31.      * @throws IOException
  32.      */
  33.     public boolean isTableExist(String tableName) throws IOException {
  34.         boolean result = false;
  35.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  36.         try {
  37.             result = admin.tableExists(tableName);
  38.         } catch (Exception e) {
  39.             log.error(e.getMessage());
  40.         } finally {
  41.             admin.close();
  42.         }
  43.         return result;
  44.     }
  46.     /**
  47.      * 创建表（DDL）
  48.      * create 't1', 'cf1', 'cf2'
  49.      *
  50.      * @param tableName    表名
  51.      * @param columnFamily 列簇
  52.      * @throws IOException
  53.      */
  54.     public void createTable(String tableName, String... columnFamily) throws IOException {
  55.         // 判断表是否存在
  56.         if (isTableExist(tableName)) {
  57.             log.error("Table:" + tableName + " already exists!");
  58.             return;
  59.         }
  61.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  62.         try {
  63.             // 创建表属性对象,表名需要转字节
  64.             HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
  65.             // 创建多个列族
  66.             for (String cf : columnFamily) {
  67.                 descriptor.addFamily(new HColumnDescriptor(cf));
  68.             }
  69.             // 根据对表的配置，创建表
  70.             admin.createTable(descriptor);
  71.             log.info("Table:" + tableName + " create successfully!");
  72.         } catch (Exception e) {
  73.             log.error(e.getMessage());
  74.         } finally {
  75.             admin.close();
  76.         }
  77.     }
  79.     /**
  80.      * 新增记录（DML）
  81.      * put 't1', '1001', 'cf1:name', 'zhangsan'
  82.      * put 't1', '1001', 'cf1:age', '23'
  83.      *
  84.      * @param tableName    表名
  85.      * @param rowKey       行键
  86.      * @param columnFamily 列簇
  87.      * @param column       列
  88.      * @param value        值
  89.      */
  90.     public void addRowData(String tableName, String rowKey, String columnFamily, String
  91.             column, String value) {
  92.         hbaseTemplate.put(tableName, rowKey, columnFamily, column, Bytes.toBytes(value));
  93.     }
  95.     /**
  96.      * 全表扫描（Scan）
  97.      * scan 't1'
  98.      * [
  99.      * {cf2={name=zhangsan, age=23}, cf1={name=tom1, age=23}},
  100.      * {cf2={name=lisi, age=24}}
  101.      * ]
  102.      *
  103.      * @param tableName 表名
  104.      * @return
  105.      */
  106.     public List<Map<String, Map<String, String>>> getAllRows(String tableName) {
  107.         Scan scan = new Scan();
  108.         List<Map<String, Map<String, String>>> list = hbaseTemplate.find(tableName, scan, (result, rowNum) -> {
  109.             Cell[] cells = result.rawCells();
  110.             Map<String, Map<String, String>> data = new HashMap<>(16);
  111.             for (Cell c : cells) {
  112.                 String columnFamily = new String(CellUtil.cloneFamily(c));
  113.                 String rowName = new String(CellUtil.cloneQualifier(c));
  114.                 String value = new String(CellUtil.cloneValue(c));
  115.                 Map<String, String> obj = data.get(columnFamily);
  116.                 if (null == obj) {
  117.                     obj = new HashMap<>(16);
  118.                 }
  119.                 obj.put(rowName, value);
  120.                 data.put(columnFamily, obj);
  121.             }
  122.             return data;
  123.         });
  124.         return list;
  125.     }
  127.     /**
  128.      * 根据RowKey获取单行记录（get）
  129.      * get 't1','1001'
  130.      * {cf1_name=tom, cf1_age=23}
  131.      *
  132.      * @param tableName 表名
  133.      * @param rowKey    行键
  134.      * @return
  135.      */
  136.     public Map<String, Map<String, Object>> getRowByRowKey(String tableName, String rowKey) {
  137.         return hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String, Map<String, Object>>>() {
  138.             @Override
  139.             public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  140.                 Cell[] cells = result.rawCells();
  141.                 Map<String, Map<String, Object>> data = new HashMap<>(16);
  142.                 for (Cell c : cells) {
  143.                     String columnFamily = new String(CellUtil.cloneFamily(c));
  144.                     String rowName = new String(CellUtil.cloneQualifier(c));
  145.                     String value = new String(CellUtil.cloneValue(c));
  146.                     Map<String, Object> obj = data.get(columnFamily);
  147.                     if (null == obj) {
  148.                         obj = new HashMap<>(16);
  149.                     }
  150.                     obj.put(rowName, value);
  151.                     data.put(columnFamily, obj);
  152.                 }
  153.                 return data;
  154.             }
  155.         });
  156.     }
  158.     /**
  159.      * 获取多行记录（get）
  160.      *
  161.      * @param tableName 表名
  162.      * @param rowKeys   行键（可扩展参数）
  163.      * @return
  164.      */
  165.     public List<Map<String, Map<String, Object>>> getRows(String tableName, String... rowKeys) {
  166.         List<Map<String, Map<String, Object>>> rows = new ArrayList<>();
  167.         for (String rowKey : rowKeys) {
  168.             Map<String, Map<String, Object>> row = hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String, Map<String, Object>>>() {
  169.                 @Override
  170.                 public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  171.                     Cell[] cells = result.rawCells();
  172.                     Map<String, Map<String, Object>> data = new HashMap<>(16);
  173.                     for (Cell c : cells) {
  174.                         String columnFamily = new String(CellUtil.cloneFamily(c));
  175.                         String rowName = new String(CellUtil.cloneQualifier(c));
  176.                         String value = new String(CellUtil.cloneValue(c));
  177.                         Map<String, Object> obj = data.get(columnFamily);
  178.                         if (null == obj) {
  179.                             obj = new HashMap<>(16);
  180.                         }
  181.                         obj.put(rowName, value);
  182.                         obj.put("rowKey", rowKey);
  183.                         data.put(columnFamily, obj);
  184.                     }
  185.                     return data;
  186.                 }
  187.             });
  188.             rows.add(row);
  189.         }
  190.         return rows;
  191.     }
  193.     /**
  194.      * 根据限定符获取指定数据（get）
  195.      *
  196.      * @param tableName  表名
  197.      * @param rowName    行键
  198.      * @param familyName 列簇
  199.      * @param qualifier  限定符
  200.      * @return
  201.      */
  202.     public String getRowQualifier(String tableName, String rowName, String familyName, String qualifier) {
  203.         return hbaseTemplate.get(tableName, rowName, familyName, qualifier, (result, i) -> {
  204.             List<Cell> ceList = result.listCells();
  205.             String res = "";
  206.             if (ceList != null && ceList.size() > 0) {
  207.                 for (Cell cell : ceList) {
  208.                     res = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  209.                 }
  210.             }
  211.             return res;
  212.         });
  213.     }
  215.     /**
  216.      * 根据StartKey和EndKey获取数据（scan），前闭后开区间
  217.      *
  218.      * @param tableName 表名
  219.      * @param startRow  起始rowKey
  220.      * @param stopRow   结束rowKey
  221.      * @return
  222.      */
  223.     public List<Map<String, Map<String, Object>>> getRowsByKey(String tableName, String startRow, String stopRow) {
  224.         Scan scan = new Scan();
  225.         if (startRow == null) {
  226.             startRow = "";
  227.         }
  228.         if (stopRow == null) {
  229.             stopRow = "";
  230.         } else {
  231.             stopRow += "|";
  232.         }
  233.         scan.setStartRow(Bytes.toBytes(startRow));
  234.         scan.setStopRow(Bytes.toBytes(stopRow));
  235.         /* PageFilter filter = new PageFilter(5);
  236.          scan.setFilter(filter);*/
  237.         return hbaseTemplate.find(tableName, scan, new RowMapper<Map<String, Map<String, Object>>>() {
  238.             @Override
  239.             public Map<String, Map<String, Object>> mapRow(Result result, int i) {
  240.                 List<Cell> ceList = result.listCells();
  241.                 Map<String, Map<String, Object>> data = new HashMap<>(16);
  242.                 if (ceList != null && ceList.size() > 0) {
  243.                     for (Cell cell : ceList) {
  244.                         String rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
  245.                         String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  246.                         String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  247.                         String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  248.                         Map<String, Object> obj = data.get(family);
  249.                         if (null == obj) {
  250.                             obj = new HashMap<>(16);
  251.                         }
  252.                         obj.put("rowKey", rowKey);
  253.                         obj.put(qualifier, value);
  254.                         data.put(family, obj);
  255.                     }
  256.                 }
  257.                 return data;
  258.             }
  259.         });
  260.     }
  262.     /**
  263.      * 删除列簇（DML）
  264.      *
  265.      * @param tableName  表名
  266.      * @param rowKey     行键
  267.      * @param familyName 列簇
  268.      */
  269.     public void deleteRow(String tableName, String rowKey, String familyName) {
  270.         hbaseTemplate.delete(tableName, rowKey, familyName);
  271.     }
  273.     /**
  274.      * 删除限定符（DML）
  275.      *
  276.      * @param tableName  表名
  277.      * @param rowKey     行键
  278.      * @param familyName 列簇
  279.      * @param qualifier  限定符
  280.      */
  281.     public void deleteQualifier(String tableName, String rowKey, String familyName, String qualifier) {
  282.         hbaseTemplate.delete(tableName, rowKey, familyName, qualifier);
  283.     }
  285.     /**
  286.      * 删除多行数据（DML）
  287.      *
  288.      * @param tableName   表名
  289.      * @param familyNames 列簇
  290.      * @param rowKeys     行键（可扩展参数）
  291.      */
  292.     public void deleteRows(String tableName, String[] familyNames, String... rowKeys) {
  293.         for (String rowKey : rowKeys) {
  294.             for (String familyName : familyNames) {
  295.                 hbaseTemplate.delete(tableName, rowKey, familyName);
  296.             }
  297.         }
  298.     }
  300.     /**
  301.      * 根据rowKey删除记录（DML）
  302.      *
  303.      * @param tableName 表名
  304.      * @param rowKey    行键
  305.      */
  306.     public void deleteAll(String tableName, final String rowKey) {
  307.         hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {
  308.             public Boolean doInTable(HTableInterface table) {
  309.                 boolean flag = false;
  310.                 try {
  311.                     List<Delete> list = new ArrayList<>();
  312.                     Delete d1 = new Delete(rowKey.getBytes());
  313.                     list.add(d1);
  314.                     table.delete(list);
  315.                     flag = true;
  316.                 } catch (Exception e) {
  317.                     e.printStackTrace();
  318.                 }
  319.                 return flag;
  320.             }
  321.         });
  322.     }
  324.     /**
  325.      * 删除表（DDL）
  326.      *
  327.      * @param tableName 表名
  328.      * @throws IOException
  329.      */
  330.     public void dropTable(String tableName) throws IOException {
  331.         if (!isTableExist(tableName)) {
  332.             log.error("Table:" + tableName + " not exist!");
  333.             return;
  334.         }
  335.         HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
  336.         try {
  337.             if (isTableExist(tableName)) {
  338.                 admin.disableTable(tableName);
  339.                 admin.deleteTable(tableName);
  340.                 log.info("Table:" + tableName + " delete successfully!");
  341.             }
  342.         } catch (Exception e) {
  343.             log.error(e.getMessage());
  344.         } finally {
  345.             admin.close();
  346.         }
  347.     }
  349.     /**
  350.      * 通用方法（一般可用于更新操作，如：delete、put）
  351.      示例：deleteAll
  352.      TableCallback action = new TableCallback<Boolean>() {
  353.          public Boolean doInTable(HTableInterface table) {
  354.              boolean flag = false;
  355.              try {
  356.                  List<Delete> list = new ArrayList<>();
  357.                  Delete d1 = new Delete(rowKey.getBytes());
  358.                  list.add(d1);
  359.                  table.delete(list);
  360.                  flag = true;
  361.              } catch (Exception e) {
  362.                 e.printStackTrace();
  363.              }
  364.              return flag;
  365.          }
  366.      };
  367.      示例：put
  368.      TableCallback action = new TableCallback<Boolean>() {
  369.          public Boolean doInTable(HTableInterface table) {
  370.              boolean flag = false;
  371.              try {
  372.                  Put put = new Put(rowKey.getBytes());
  373.                  put.add(Bytes.toBytes(CF_01), Bytes.toBytes(CF1_QUALIFIER_NAME), Bytes.toBytes(CF1_NAME_VALUE));
  374.                  put.add(Bytes.toBytes(CF_02), Bytes.toBytes(CF2_QUALIFIER_ADDR), Bytes.toBytes(CF2_ADDR_VALUE));
  375.                  table.put(put);
  376.                  flag = true;
  377.              } catch (Exception e) {
  378.                 e.printStackTrace();
  379.              }
  380.              return flag;
  381.          }
  382.      };
  383.      * @param tableName
  384.      * @param action
  385.      */
  386.     public void execute(String tableName, TableCallback action) {
  387.         hbaseTemplate.execute(tableName, action);
  388.     }
  390. }

4. 测试

■ 测试类（HBaseServiceApplicationTests.java）

  1. import org.apache.hadoop.hbase.client.Delete;
  2. import org.apache.hadoop.hbase.client.HTableInterface;
  3. import org.apache.hadoop.hbase.client.Put;
  4. import org.apache.hadoop.hbase.util.Bytes;
  5. import org.junit.Assert;
  6. import org.junit.jupiter.api.Test;
  7. import org.junit.runner.RunWith;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.boot.test.context.SpringBootTest;
  10. import org.springframework.data.hadoop.hbase.TableCallback;
  11. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  12. import java.io.IOException;
  13. import java.util.ArrayList;
  14. import java.util.List;
  15. import java.util.Map;
  17. @RunWith(SpringJUnit4ClassRunner.class)
  18. @SpringBootTest
  19. class HBaseServiceApplicationTests {
  21.     private String tableName = "t1";
  22.     private String rowKey = "1001";
  23.     private String rowKey2 = "1002";
  24.     private String rowKey3 = "1003";
  26.     private String CF_01 = "cf1";
  27.     private String CF1_QUALIFIER_NAME = "name";
  28.     private String CF1_QUALIFIER_AGE = "age";
  29.     private String CF1_NAME_VALUE = "tom";
  30.     private String CF1_AGE_VALUE = "23";
  33.     private String CF_02 = "cf2";
  34.     private String CF2_QUALIFIER_ADDR = "address";
  35.     private String CF2_ADDR_VALUE = "beijing";
  37.     @Autowired
  38.     private HBaseService service;
  40.     @Test
  41.     public void isTableExist() throws IOException {
  42.         assert service.isTableExist(tableName);
  43.     }
  45.     @Test
  46.     public void createTable() throws IOException {
  47.         service.createTable(tableName, CF_01, CF_02);
  48.         assert service.isTableExist(tableName);
  49.     }
  51.     @Test
  52.     public void addRowData() {
  53.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  54.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  55.     }
  57.     @Test
  58.     public void getAllRows() {
  59.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  60.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  61.         List<Map<String, Map<String, String>>> rows = service.getAllRows(tableName);
  62.         Assert.assertTrue(rows.size() == 1); // 1个RowKey，对应1行数据
  63.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  64.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  65.         Assert.assertEquals(CF1_AGE_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_AGE));
  66.         System.out.println(rows);
  67.     }
  69.     @Test
  70.     public void getRowByRowKey() {
  71.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  72.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  73.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  74.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  75.         Assert.assertNotNull(row);
  76.         Assert.assertEquals(CF1_NAME_VALUE, row.get(CF_01).get(CF1_QUALIFIER_NAME));
  77.         Assert.assertEquals(CF1_AGE_VALUE, row.get(CF_01).get(CF1_QUALIFIER_AGE));
  78.         Assert.assertEquals(CF2_ADDR_VALUE, row.get(CF_02).get(CF2_QUALIFIER_ADDR));
  79.         System.out.println(row);
  80.     }
  82.     @Test
  83.     public void getRows() {
  84.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  85.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  86.         List<Map<String, Map<String, Object>>> rows = service.getRows(tableName, rowKey2, rowKey3);
  87.         Assert.assertTrue(rows.size() == 2); // 2个RowKey，对应1行数据
  88.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  89.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  90.         System.out.println(rows);
  91.     }
  93.     @Test
  94.     public void getRowQualifier() {
  95.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  96.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  97.     }
  99.     @Test
  100.     public void getRowsByKey() {
  101.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  102.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  103.         List<Map<String, Map<String, Object>>> rows = service.getRowsByKey(tableName, rowKey2, rowKey3);
  104.         Assert.assertTrue(rows.size() == 2); // 2个RowKey，对应1行数据
  105.         Assert.assertTrue(rows.get(0).size() == 1); // 1个列簇
  106.         Assert.assertEquals(CF1_NAME_VALUE, rows.get(0).get(CF_01).get(CF1_QUALIFIER_NAME));
  107.         System.out.println(rows);
  108.     }
  110.     @Test
  111.     public void deleteQualifier() {
  112.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  113.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_AGE, CF1_AGE_VALUE);
  114.         service.deleteQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME);
  115.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  116.         System.out.println(row);
  117.         Assert.assertNull(row.get(CF_01).get(CF1_QUALIFIER_NAME));
  118.         Assert.assertNotNull(row.get(CF_01).get(CF1_QUALIFIER_AGE));
  119.     }
  121.     @Test
  122.     public void deleteRow() {
  123.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  124.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  125.         service.deleteRow(tableName, rowKey, CF_01);
  126.         service.deleteRow(tableName, rowKey, CF_02);
  127.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  128.         System.out.println(row);
  129.         Assert.assertTrue(row.isEmpty());
  130.     }
  132.     @Test
  133.     public void deleteMultiRow() {
  134.         service.addRowData(tableName, rowKey2, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  135.         service.addRowData(tableName, rowKey2, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  136.         service.addRowData(tableName, rowKey3, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  137.         service.addRowData(tableName, rowKey3, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  138.         List<Map<String, Map<String, Object>>> rows = service.getRows(tableName, rowKey2, rowKey3);
  139.         System.out.println("----------------------- test 1 -------------------------");
  140.         System.out.println(rows);
  142.         String[] familyNames = {CF_01, CF_02};
  143.         service.deleteRows(tableName, familyNames, rowKey2, rowKey3);
  144.         rows = service.getRows(tableName, rowKey2, rowKey3);
  145.         System.out.println("----------------------- test 2 -------------------------");
  146.         System.out.println(rows); // [{}, {}]
  147.         Assert.assertTrue(rows.get(0).isEmpty());
  148.     }
  150.     @Test
  151.     public void deleteAll() {
  152.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  153.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  154.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  155.         System.out.println(row);
  156.         service.deleteAll(tableName, rowKey);
  157.         row = service.getRowByRowKey(tableName, rowKey);
  158.         System.out.println(row);
  159.         Assert.assertTrue(row.isEmpty());
  160.     }
  162.     @Test
  163.     public void dropTable() throws IOException {
  164.         if (!service.isTableExist(tableName)) {
  165.             service.createTable(tableName);
  166.         }
  167.         assert service.isTableExist(tableName);
  168.         service.dropTable(tableName);
  169.         assert !service.isTableExist(tableName);
  170.     }
  172.     @Test
  173.     public void execute_deleteAll() {
  174.         service.addRowData(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME, CF1_NAME_VALUE);
  175.         service.addRowData(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR, CF2_ADDR_VALUE);
  176.         Map<String, Map<String, Object>> row = service.getRowByRowKey(tableName, rowKey);
  177.         System.out.println(row);
  178.         TableCallback action = new TableCallback<Boolean>() {
  179.             public Boolean doInTable(HTableInterface table) {
  180.                 boolean flag = false;
  181.                 try {
  182.                     List<Delete> list = new ArrayList<>();
  183.                     Delete d1 = new Delete(rowKey.getBytes());
  184.                     list.add(d1);
  185.                     table.delete(list);
  186.                     flag = true;
  187.                 } catch (Exception e) {
  188.                     e.printStackTrace();
  189.                 }
  190.                 return flag;
  191.             }
  192.         };
  193.         service.execute(tableName, action);
  194.         row = service.getRowByRowKey(tableName, rowKey);
  195.         System.out.println(row);
  196.         Assert.assertTrue(row.isEmpty());
  197.     }
  199.     @Test
  200.     public void execute_put() {
  201.         TableCallback action = new TableCallback<Boolean>() {
  202.             public Boolean doInTable(HTableInterface table) {
  203.                 boolean flag = false;
  204.                 try {
  205.                     Put put = new Put(rowKey.getBytes());
  206.                     put.add(Bytes.toBytes(CF_01), Bytes.toBytes(CF1_QUALIFIER_NAME), Bytes.toBytes(CF1_NAME_VALUE));
  207.                     put.add(Bytes.toBytes(CF_02), Bytes.toBytes(CF2_QUALIFIER_ADDR), Bytes.toBytes(CF2_ADDR_VALUE));
  208.                     table.put(put);
  209.                     flag = true;
  210.                 } catch (Exception e) {
  211.                     e.printStackTrace();
  212.                 }
  213.                 return flag;
  214.             }
  215.         };
  216.         service.execute(tableName, action);
  217.         Assert.assertEquals(CF1_NAME_VALUE, service.getRowQualifier(tableName, rowKey, CF_01, CF1_QUALIFIER_NAME));
  218.         Assert.assertEquals(CF2_ADDR_VALUE, service.getRowQualifier(tableName, rowKey, CF_02, CF2_QUALIFIER_ADDR));
  219.     }
  221. }

■ Runner测试类（HBaseServiceRunnerTest.java）

  1. import com.lonton.bigdata.service.HBaseService;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.ApplicationArguments;
  6. import org.springframework.boot.ApplicationRunner;
  7. import org.springframework.stereotype.Component;
  9. @Component
  10. public class HBaseServiceRunnerTest implements ApplicationRunner {
  12.     private final static Logger log = LoggerFactory.getLogger(HBaseServiceRunnerTest.class);
  14.     private String tableName = "t1";
  16.     @Autowired
  17.     private HBaseService service;
  19.     @Override
  20.     public void run(ApplicationArguments args) throws Exception {
  21.         log.info("------------------------- HBaseServiceTest ApplicationRunner --------------------------------");
  22.         log.info(String.valueOf(service.isTableExist(tableName)));
  23.     }
  25. }

5. 完整示例代码

源码：hbase-springboot-example04-src.zip

方式5：SpringBoot + hbase starter

spring-boot-starter-hbase是spring-boot自定义的starter，为hbase的query和更新等操作提供简易的api并集成spring-boot的auto configuration。如果HbaseTemplate操作不满足需求，完全可以使用hbaseTemplate的getConnection()方法，获取连接。进而类似HbaseTemplate实现的逻辑，实现更复杂的需求查询等功能**。**
源码：https://github.com/SpringForAll/spring-boot-starter-hbase
示例：https://github.com/JeffLi1993/springboot-learning-example
参考：Spring Boot 2.x：通过spring-boot-starter-hbase集成HBase

1. 添加pom.xml依赖

  1. <!-- Spring Boot HBase 依赖 -->
  2. <!-- https://mvnrepository.com/artifact/com.spring4all/spring-boot-starter-hbase -->
  3. <dependency>
  4.     <groupId>com.spring4all</groupId>
  5.     <artifactId>spring-boot-starter-hbase</artifactId>
  6.     <version>1.0.0.RELEASE</version>
  7. </dependency>

补充：安装spring-boot-starter-hbase组件依赖（因为不在公共仓库，只能自行安装。如果有maven私库，可以考虑安装到私库）

  1. # 下载项目到本地
  2. git clone https://github.com/SpringForAll/spring-boot-starter-hbase.git
  3. cd spring-boot-starter-hbase
  4. # 安装依赖
  5. mvn clean install

源码：spring-boot-starter-hbase-master.zip

2. 工程配置

■ SpringBoot配置（application.properties）

  1. ## HBase Configuration
  2. spring.data.hbase.quorum=192.168.0.101:2181,192.168.0.102:2181,192.168.0.103:2181
  3. spring.data.hbase.rootDir=hdfs://bigdata-node1:9000/hbase
  4. spring.data.hbase.nodeParent=/hbase

具体配置项信息如下：

  • spring.data.hbase.quorum：指定HBase的zk地址
  • spring.data.hbase.rootDir：指定HBase在HDFS上存储的路径
  • spring.data.hbase.nodeParent：指定ZK中HBase的根ZNode

    3. 编写关键类

    ■ Service接口（HBaseService.java） ```java import com.spring4all.spring.boot.starter.hbase.api.TableCallback; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.Mutation; import org.apache.hadoop.hbase.client.Scan; import java.util.List;

/**

  • Description HBase服务接口 */ public interface HBaseService {

    /**

    • 根据rowKey获取单行记录，数据由mapper映射 *
    • @param tableName 表名
    • @param rowKey 行键
    • @param 泛型

    • @return */ T get(String tableName, String rowKey);

      /**

    • 根据rowKey和列簇取单行记录，数据由mapper映射 *
    • @param tableName 表名
    • @param rowKey 行键
    • @param familyName 列簇
    • @param 泛型

    • @return */ T get(String tableName, String rowKey, String familyName);

      /**

    • 根据限定符获取单行记录，数据由mapper映射 *
    • @param tableName 表名
    • @param rowKey 行键
    • @param familyName 列簇
    • @param qualifier 限定符
    • @param 泛型

    • @return */ T get(String tableName, String rowKey, String familyName, String qualifier);

      /**

    • 新增或修改
    • 注意：写入非字符串型的数据时，要先转为String，如：int，Bytes.toBytes(String.valueOf(值)) *
    • @param tableName 表名

    • @param mutation 执行put update or delete */ void saveOrUpdate(String tableName, Mutation mutation);

      /**

    • 批量新增或修改
    • 注意：写入非字符串型的数据时，要先转为String，如：int，Bytes.toBytes(String.valueOf(值)) *
    • @param tableName 表名

    • @param mutations 批量执行put update or delete */ void saveOrUpdates(String tableName, List mutations);

      /**

    • 通用方法（一般可用于更新操作，如：delete、put）
      // 执行操作（deleteAll） TableCallback action = new TableCallback() {
      1.  @Override
      2.  public Boolean doInTable(Table table) throws Throwable {
      3.      boolean flag = false;
      4.      try {
      5.          List<Delete> list = new ArrayList<>();
      6.          Delete d1 = new Delete(rowKey.getBytes());
      7.          list.add(d1);
      8.          table.delete(list);
      9.          flag = true;
      10.      } catch (Exception e) {
      11.          e.printStackTrace();
      12.      }
      13.      return flag;
      14.  }
      }; // 执行操作（put update） TableCallback action = new TableCallback() {
      1.  @Override
      2.  public Boolean doInTable(Table table) throws Throwable {
      3.      boolean flag = false;
      4.      try {
      5.          Put put = new Put(rowKey.getBytes());
      6.          put.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
      7.          put.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
      8.          put.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
      9.          table.put(put);
      10.          flag = true;
      11.      } catch (Exception e) {
      12.          e.printStackTrace();
      13.      }
      14.      return flag;
      15.  }
      };
    • @param tableName 表名
    • @param action 执行put update or delete
    • @param 泛型

    • @return */ T execute(String tableName, TableCallback action);

      /**

    • 扫描 *
    • @param tableName 表名
    • @param scan 扫描对象
    • @param 泛型

    • @return */ List find(String tableName, Scan scan);

      /**

    • 扫描（根据列簇）
    • @param tableName
    • @param family
    • @param

    • @return */ List find(String tableName, String family);

      /**

    • 扫描（根据限定符）
    • @param tableName
    • @param family
    • @param qualifier
    • @param

    • @return */ List find(String tableName, String family, String qualifier);

      /**

    • 如果HbaseTemplate操作不满足需求，可以使用hbaseTemplate的getConnection()方法，
    • 获取连接，进而类似HbaseTemplate实现的逻辑，实现更复杂的需求查询等功能

    • @return */ Connection getConnection();

      /**

    • 关闭连接
    • @param connection */ void closeConnection(Connection connection);

}

  1. <a name="Bi7lI"></a>
  2. ## 4. 测试
  3. **■ DTO定义（Student.java）**
  4. ```java
  5. import lombok.AllArgsConstructor;
  6. import lombok.Data;
  7. import lombok.NoArgsConstructor;
  9. /**
  10.  * Description DTO-学生实体类
  11.  * create 'student','base','other'
  12.  */
  13. @Data
  14. @NoArgsConstructor
  15. @AllArgsConstructor
  16. public class Student {
  18.     private BaseInfo base;
  19.     private OtherInfo other;
  21.     @Data
  22.     @NoArgsConstructor
  23.     @AllArgsConstructor
  24.     public static class BaseInfo{
  25.         private String name;
  26.         private Integer age;
  28.         @Override
  29.         public String toString() {
  30.             return "BaseInfo{" +
  31.                     "name='" + name + '\'' +
  32.                     ", age=" + age +
  33.                     '}';
  34.         }
  35.     }
  37.     @Data
  38.     @NoArgsConstructor
  39.     @AllArgsConstructor
  40.     public static class OtherInfo{
  41.         private Integer fv;
  43.         @Override
  44.         public String toString() {
  45.             return "OtherInfo{" +
  46.                     "fv=" + fv +
  47.                     '}';
  48.         }
  49.     }
  52.     @Override
  53.     public String toString() {
  54.         return "Student{" +
  55.                 "base=" + base +
  56.                 ", other=" + other +
  57.                 '}';
  58.     }
  60. }

■ RowMapper定义（StudentRowMapper.java）

  1. import com.lonton.bigdata.domain.Student;
  2. import com.spring4all.spring.boot.starter.hbase.api.RowMapper;
  3. import org.apache.hadoop.hbase.client.Result;
  4. import org.apache.hadoop.hbase.util.Bytes;
  6. /**
  7.  * Description HBase Mapper
  8.  */
  9. public class StudentRowMapper implements RowMapper<Student> {
  11.     private static final byte[] BASE_FAMILY = "base".getBytes();
  12.     private static final byte[] OTHER_FAMILY = "other".getBytes();
  13.     private static final byte[] NAME = "name".getBytes();
  14.     private static final byte[] AGE = "age".getBytes();
  15.     private static final byte[] FV = "fv".getBytes();
  17.     /*方法1：ok*/
  18.     /*
  19.     @Override
  20.     public Student mapRow(Result result, int rowNum) throws Exception {
  21.         Student.BaseInfo baseInfo = new Student.BaseInfo();
  22.         Student.OtherInfo otherInfo = new Student.OtherInfo();
  24.         Cell[] cells = result.rawCells();
  25.         for (Cell c : cells) {
  26.             String columnFamily = new String(CellUtil.cloneFamily(c));
  27.             String rowName = new String(CellUtil.cloneQualifier(c));
  28.             String value = new String(CellUtil.cloneValue(c));
  29.             if(value!=null){
  30.                 if(Bytes.toString(BASE_FAMILY).equals(columnFamily)){
  31.                     if(Bytes.toString(NAME).equals(rowName)){
  32.                         baseInfo.setName(value);
  33.                     }else if(Bytes.toString(AGE).equals(rowName)){
  34.                         Integer age=Integer.parseInt(value);
  35.                         baseInfo.setAge(age);
  36.                     }
  37.                 } else if(Bytes.toString(OTHER_FAMILY).equals(columnFamily)){
  38.                     if(Bytes.toString(FV).equals(rowName)){
  39.                         Integer fv=Integer.parseInt(value);
  40.                         otherInfo.setFv(fv);
  41.                     }
  42.                 }
  43.             }
  44.         }
  46.         Student dto = new Student(baseInfo, otherInfo);
  47.         return dto;
  48.     }*/
  51.     /*方法2：ok*/
  52.     @Override
  53.     public Student mapRow(Result result, int rowNum) throws Exception {
  54.         Student.BaseInfo baseInfo = new Student.BaseInfo();
  55.         if (result.containsColumn(BASE_FAMILY, NAME)) {
  56.             String name = Bytes.toString(result.getValue(BASE_FAMILY, NAME));
  57.             baseInfo.setName(name);
  58.         }
  59.         if (result.containsColumn(BASE_FAMILY, AGE)) {
  60.             Integer age = Integer.parseInt(Bytes.toString(result.getValue(BASE_FAMILY, AGE)));
  61.             baseInfo.setAge(age);
  62.         }
  64.         Student.OtherInfo otherInfo = new Student.OtherInfo();
  65.         if (result.containsColumn(OTHER_FAMILY, FV)) {
  66.             Integer fv = Integer.parseInt(Bytes.toString(result.getValue(OTHER_FAMILY, FV)));
  67.             otherInfo.setFv(fv);
  68.         }
  70.         Student dto = new Student(baseInfo, otherInfo);
  71.         return dto;
  72.     }
  74. }

■ Service实现类（StudentHBaseServiceImpl.java）

  1. import com.lonton.bigdata.domain.Student;
  2. import com.lonton.bigdata.mapper.StudentRowMapper;
  3. import com.lonton.bigdata.service.HBaseService;
  4. import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
  5. import com.spring4all.spring.boot.starter.hbase.api.TableCallback;
  6. import org.apache.hadoop.hbase.client.Connection;
  7. import org.apache.hadoop.hbase.client.Mutation;
  8. import org.apache.hadoop.hbase.client.Scan;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Service;
  11. import java.io.IOException;
  12. import java.util.List;
  14. /**
  15.  * Description HBase服务接口实现类
  16.  */
  17. @Service
  18. public class StudentHBaseServiceImpl implements HBaseService {
  20.     @Autowired
  21.     private HbaseTemplate hbaseTemplate;
  23.     @Override
  24.     public Student get(String tableName, String rowKey) {
  25.         return hbaseTemplate.get(tableName,rowKey,new StudentRowMapper());
  26.     }
  28.     @Override
  29.     public Student get(String tableName, String rowKey, String familyName) {
  30.         return hbaseTemplate.get(tableName,rowKey,familyName,new StudentRowMapper());
  31.     }
  33.     @Override
  34.     public Student get(String tableName, String rowKey, String familyName, String qualifier) {
  35.         return hbaseTemplate.get(tableName,rowKey,familyName,qualifier,new StudentRowMapper());
  36.     }
  38.     @Override
  39.     public void saveOrUpdate(String tableName, Mutation mutation) {
  40.         hbaseTemplate.saveOrUpdate(tableName,mutation);
  41.     }
  43.     @Override
  44.     public void saveOrUpdates(String tableName, List<Mutation> mutations) {
  45.         hbaseTemplate.saveOrUpdates(tableName,mutations);
  46.     }
  48.     @Override
  49.     public <T> T execute(String tableName, TableCallback<T> action) {
  50.         return hbaseTemplate.execute(tableName,action);
  51.     }
  53.     @Override
  54.     public List<Student> find(String tableName, Scan scan) {
  55.         return hbaseTemplate.find(tableName,scan,new StudentRowMapper());
  56.     }
  58.     @Override
  59.     public List<Student> find(String tableName, String family) {
  60.         return hbaseTemplate.find(tableName,family,new StudentRowMapper());
  61.     }
  63.     @Override
  64.     public List<Student> find(String tableName, String family, String qualifier) {
  65.         return hbaseTemplate.find(tableName,family,qualifier,new StudentRowMapper());
  66.     }
  68.     @Override
  69.     public Connection getConnection() {
  70.         return hbaseTemplate.getConnection();
  71.     }
  73.     @Override
  74.     public void closeConnection(Connection connection) {
  75.         if(connection!=null){
  76.             try {
  77.                 connection.close();
  78.             } catch (IOException e) {
  79.                 e.printStackTrace();
  80.             }
  81.         }
  82.     }
  84. }

■ 测试类（HBaseServiceApplicationTests.java）

  1. import com.lonton.bigdata.domain.Student;
  2. import com.spring4all.spring.boot.starter.hbase.api.TableCallback;
  3. import org.apache.hadoop.hbase.client.*;
  4. import org.apache.hadoop.hbase.util.Bytes;
  5. import org.junit.Assert;
  6. import org.junit.jupiter.api.Test;
  7. import org.junit.runner.RunWith;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.boot.test.context.SpringBootTest;
  12. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  13. import java.util.ArrayList;
  14. import java.util.List;
  15. import java.util.Map;
  17. /****
  18.  * @Description HBase单元测试类
  19.  * create 'student','base','other'
  20.  * truncate 'student'
  21.  * put 'student','1001','base:name','zhangsan'
  22.  * put 'student','1001','base:age',23
  23.  * put 'student','1001','other:fv',3
  24.  * scan 'student'
  25.  */
  26. @RunWith(SpringJUnit4ClassRunner.class)
  27. @SpringBootTest
  28. class HBaseServiceApplicationTests {
  30.     private String tableName = "student";
  31.     private String rowKey = "1001";
  33.     private String CF_BASE = "base";
  34.     private String CF_OTHER = "other";
  36.     private String CF_BASE_QUALIFIER_NAME = "name";
  37.     private String CF_BASE_QUALIFIER_AGE = "age";
  38.     private String CF_OTHER_QUALIFIER_FV = "fv";
  40.     private String CF_BASE_VALUE_NAME = "zhangsan";
  41.     private Integer CF_BASE_VALUE_AGE = 23;
  42.     private Integer CF_OTHER_VALUE_FV = 3;
  44.     @Autowired
  45.     private HBaseService service;
  47.     /** Student{base=BaseInfo{name='zhangsan', age=23}, other=OtherInfo{fv=3}} **/
  48.     @Test
  49.     public void get1() {
  50.         // 新增测试数据
  51.         Put mutation = new Put(Bytes.toBytes(rowKey));
  52.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  53.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  54.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  55.         service.saveOrUpdate(tableName, mutation);
  56.         // 获取记录
  57.         Student dto=service.get(tableName,rowKey);
  58.         // 验证
  59.         System.out.println(dto);
  60.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  61.         Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  62.         Assert.assertEquals(CF_OTHER_VALUE_FV, dto.getOther().getFv());
  63.     }
  65.     /** Student{base=BaseInfo{name='zhangsan', age=23}, other=OtherInfo{fv=null}} **/
  66.     @Test
  67.     public void get2() {
  68.         // 新增测试数据
  69.         Put mutation = new Put(Bytes.toBytes(rowKey));
  70.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  71.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  72.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  73.         service.saveOrUpdate(tableName, mutation);
  74.         // 获取记录
  75.         Student dto=service.get(tableName,rowKey,CF_BASE);
  76.         // 验证
  77.         System.out.println(dto);
  78.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  79.         Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  80.         Assert.assertNull(dto.getOther().getFv());
  81.     }
  83.     /** Student{base=BaseInfo{name='zhangsan', age=null}, other=OtherInfo{fv=null}} **/
  84.     @Test
  85.     public void get3() {
  86.         // 新增测试数据
  87.         Put mutation = new Put(Bytes.toBytes(rowKey));
  88.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  89.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  90.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  91.         service.saveOrUpdate(tableName, mutation);
  92.         // 获取记录
  93.         Student dto=service.get(tableName,rowKey,CF_BASE,CF_BASE_QUALIFIER_NAME);
  94.         // 验证
  95.         System.out.println(dto);
  96.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  97.         Assert.assertNull(dto.getBase().getAge());
  98.         Assert.assertNull(dto.getOther().getFv());
  99.     }
  101.     @Test
  102.     public void saveOrUpdate() {
  103.         // 新增测试数据
  104.         Put mutation = new Put(Bytes.toBytes(rowKey));
  105.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  106.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  107.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  108.         service.saveOrUpdate(tableName, mutation);
  109.         // 验证
  110.         Student dto=service.get(tableName,rowKey);
  111.         System.out.println(dto);
  112.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  113.         Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  114.         Assert.assertEquals(CF_OTHER_VALUE_FV, dto.getOther().getFv());
  115.     }
  117.     @Test
  118.     public void saveOrUpdates() {
  119.         List<Mutation> mutations=new ArrayList<>();
  120.         // 新增${CF_BASE_QUALIFIER_NAME}
  121.         Put mutation = new Put(Bytes.toBytes(rowKey));
  122.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  123.         mutations.add(mutation);
  124.         // 新增${CF_BASE_QUALIFIER_AGE}
  125.         mutation = new Put(Bytes.toBytes(rowKey));
  126.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  127.         mutations.add(mutation);
  128.         // 新增${CF_OTHER_QUALIFIER_FV}
  129.         mutation = new Put(Bytes.toBytes(rowKey));
  130.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  131.         mutations.add(mutation);
  132.         // 批量新增测试数据
  133.         service.saveOrUpdates(tableName, mutations);
  134.         // 验证
  135.         Student dto=service.get(tableName,rowKey);
  136.         System.out.println(dto);
  137.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  138.         Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  139.         Assert.assertEquals(CF_OTHER_VALUE_FV, dto.getOther().getFv());
  140.     }
  143.     @Test
  144.     public void execute_deleteAll() {
  145.         // 新增测试数据
  146.         Put mutation = new Put(Bytes.toBytes(rowKey));
  147.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  148.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  149.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  150.         service.saveOrUpdate(tableName, mutation);
  151.         // 获取记录
  152.         Student dto=service.get(tableName,rowKey,CF_BASE,CF_BASE_QUALIFIER_NAME);
  153.         // 验证
  154.         System.out.println(dto);
  155.         // 执行操作（deleteAll）
  156.         TableCallback action = new TableCallback<Boolean>() {
  157.             @Override
  158.             public Boolean doInTable(Table table) throws Throwable {
  159.                 boolean flag = false;
  160.                 try {
  161.                     List<Delete> list = new ArrayList<>();
  162.                     Delete d1 = new Delete(rowKey.getBytes());
  163.                     list.add(d1);
  164.                     table.delete(list);
  165.                     flag = true;
  166.                 } catch (Exception e) {
  167.                     e.printStackTrace();
  168.                 }
  169.                 return flag;
  170.             }
  171.         };
  172.         service.execute(tableName, action);
  173.         // 验证
  174.         dto=service.get(tableName,rowKey);
  175.         System.out.println(dto);
  176.         Assert.assertNull(dto.getBase().getName());
  177.         Assert.assertNull(dto.getBase().getAge());
  178.         Assert.assertNull(dto.getOther().getFv());
  179.     }
  181.     @Test
  182.     public void execute_put() {
  183.         // 执行操作（put update）
  184.         TableCallback action = new TableCallback<Boolean>() {
  185.             @Override
  186.             public Boolean doInTable(Table table) throws Throwable {
  187.                 boolean flag = false;
  188.                 try {
  189.                     Put put = new Put(rowKey.getBytes());
  190.                     put.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  191.                     put.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  192.                     put.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  193.                     table.put(put);
  194.                     flag = true;
  195.                 } catch (Exception e) {
  196.                     e.printStackTrace();
  197.                 }
  198.                 return flag;
  199.             }
  200.         };
  201.         service.execute(tableName, action);
  202.         // 获取记录
  203.         Student dto=service.get(tableName,rowKey);
  204.         // 验证
  205.         System.out.println(dto);
  206.         Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  207.         Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  208.         Assert.assertEquals(CF_OTHER_VALUE_FV, dto.getOther().getFv());
  209.     }
  211.     @Test
  212.     public void find_scan() {
  213.         // 新增测试数据
  214.         String startRow=rowKey+"-01";
  215.         String stopRow=rowKey+"-02";
  216.         Put mutation = new Put(Bytes.toBytes(startRow));
  217.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  218.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  219.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  220.         service.saveOrUpdate(tableName, mutation);
  221.         mutation = new Put(Bytes.toBytes(stopRow));
  222.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  223.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  224.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  225.         service.saveOrUpdate(tableName, mutation);
  226.         // 范围扫描（scan为空则为全表扫描）
  227.         Scan scan = new Scan();
  228.         if (startRow == null) {
  229.             startRow = "";
  230.         }
  231.         if (stopRow == null) {
  232.             stopRow = "";
  233.         } else {
  234.             stopRow += "|";
  235.         }
  236.         scan.setStartRow(Bytes.toBytes(startRow));
  237.         scan.setStopRow(Bytes.toBytes(stopRow));
  238.         scan.setCaching(5000);
  239.         List<Student> dtoList=service.find(tableName,scan);
  240.         // 验证
  241.         for (Student dto : dtoList) {
  242.             System.out.println(dto);
  243.             Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  244.             Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  245.             Assert.assertEquals(CF_OTHER_VALUE_FV, dto.getOther().getFv());
  246.         }
  247.     }
  249.     @Test
  250.     public void find_family() {
  251.         // 新增测试数据
  252.         String startRow=rowKey+"-01";
  253.         String stopRow=rowKey+"-02";
  254.         Put mutation = new Put(Bytes.toBytes(startRow));
  255.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  256.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  257.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  258.         service.saveOrUpdate(tableName, mutation);
  259.         mutation = new Put(Bytes.toBytes(stopRow));
  260.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  261.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  262.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  263.         service.saveOrUpdate(tableName, mutation);
  264.         // 范围扫描
  265.         List<Student> dtoList=service.find(tableName,CF_BASE);
  266.         // 验证
  267.         for (Student dto : dtoList) {
  268.             System.out.println(dto);
  269.             Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  270.             Assert.assertEquals(CF_BASE_VALUE_AGE, dto.getBase().getAge());
  271.             Assert.assertNull(dto.getOther().getFv());
  272.         }
  273.     }
  275.     @Test
  276.     public void find_qualifier() {
  277.         // 新增测试数据
  278.         String startRow=rowKey+"-01";
  279.         String stopRow=rowKey+"-02";
  280.         Put mutation = new Put(Bytes.toBytes(startRow));
  281.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  282.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  283.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  284.         service.saveOrUpdate(tableName, mutation);
  285.         mutation = new Put(Bytes.toBytes(stopRow));
  286.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_NAME), Bytes.toBytes(CF_BASE_VALUE_NAME));
  287.         mutation.addColumn(Bytes.toBytes(CF_BASE), Bytes.toBytes(CF_BASE_QUALIFIER_AGE), Bytes.toBytes(String.valueOf(CF_BASE_VALUE_AGE)));
  288.         mutation.addColumn(Bytes.toBytes(CF_OTHER), Bytes.toBytes(CF_OTHER_QUALIFIER_FV), Bytes.toBytes(String.valueOf(CF_OTHER_VALUE_FV)));
  289.         service.saveOrUpdate(tableName, mutation);
  290.         // 范围扫描
  291.         List<Student> dtoList=service.find(tableName,CF_BASE,CF_BASE_QUALIFIER_NAME);
  292.         // 验证
  293.         for (Student dto : dtoList) {
  294.             System.out.println(dto);
  295.             Assert.assertEquals(CF_BASE_VALUE_NAME, dto.getBase().getName());
  296.             Assert.assertNull(dto.getBase().getAge());
  297.             Assert.assertNull(dto.getOther().getFv());
  298.         }
  299.     }
  301.     @Test
  302.     public void getAndCloseConnection() {
  303.         Connection connection=service.getConnection();
  304.         System.out.println(connection);
  305.         Assert.assertFalse(connection.isClosed());
  306.         service.closeConnection(connection);
  307.         Assert.assertTrue(connection.isClosed());
  308.     }
  310. }

■ 控制类（StudentController.java）

  1. import com.lonton.bigdata.domain.Student;
  2. import com.lonton.bigdata.service.HBaseService;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.stereotype.Controller;
  5. import org.springframework.web.bind.annotation.*;
  7. /**
  8.  * Description 控制类<br/>
  9.  * put 'student','1001','base:name','zhangsan'
  10.  * put 'student','1001','base:age',23
  11.  * put 'student','1001','other:fv',3
  12.  * @RestController等价于@Controller + @ResponseBody
  13.  */
  14. @RestController
  15. public class StudentController {
  17.     @Autowired
  18.     private HBaseService service;
  20.     /** http://localhost:8080/hbaseservice/student/1001 **/
  21.     /**
  22.      * {"base":{"name":"zhangsan","age":23},"other":{"fv":3}}
  23.      **/
  24.     @RequestMapping(value = "/hbaseservice/{tableName}/{rowKey}", method = RequestMethod.GET)
  25.     public Student getStudent(@PathVariable String tableName, @PathVariable String rowKey) {
  26.         return service.get(tableName, rowKey);
  27.     }
  29. }

5. 完整示例代码

源码：hbase-springboot-example05-src.zip