大数据技术(八)—— HBase数据读写流程和Api的使用

目录
一、读数据
二、写数据
1、写数据
2、数据刷到磁盘过程
3、数据合并过程
三、Api的使用
四、Phoenix
1、安装Phoenix
2、简单使用
2.1、查看表信息
2.2、创建表
2.3、插入数据
2.4、查看数据
2.5、修改数据
2.6、删除数据
2.7、退出
2.8、导入数据
3、API的使用
五、整合SpringBoot+MyBatis
六、参考

---

一、读数据

• Client先访问zookeeper，哀求获取Meta表信息。从上面图可以看到Zookeeper中存储了HBase的元数据信息。
  
• Zookeeper返回meta数据信息。
  
• 根据namespace、表名和rowkey在meta表的信息去获取数据所在的RegionServer和Region
  
• 找到这个region对应的regionserver；
  
• 查找对应的region；
  
• 先从MemStore找数据，假如没有，再到BlockCache里面读；
  
• BlockCache还没有，再到StoreFile上读(为了读取的服从)；
  
• 假如是从StoreFile里面读取的数据，不是直接返回给客户端，而是先写入BlockCache，再返回给客户端。
  

二、写数据

1、写数据

写数据和读数据前几步的过程都是一样的，首先从Zookeeper获取Meta表信息，根据meta表信息确认RegionServer 和Region

• Client向RegionServer发送写哀求。
  
• RegionServer将数据写到HLog（write ahead log）。为了数据的持久化和恢复（避免数据在内存还没刷到磁盘就宕机丢失数据）。
  
• RegionServer将数据写到内存（MemStore），此时就认为写数据完成了，至于什么时间数据写进HDFS，它不关心。
  
• 反馈Client写成功。
  

2、数据刷到磁盘过程

当MemStore数据达到阈值（默认是128M），将数据刷到硬盘，将内存中的数据删除，同时删除HLog（Write Ahead Log）中的汗青数据；并将数据存储到HDFS中；在HLog中做标记点。
3、数据合并过程

• 当数据块达到3块，Hmaster触发合并操作，Region将数据块加载到本地，进行合并。
  
• 当合并的数据超过256M，进行拆分，将拆分后的Region分配给不同的HregionServer管理。
  
• 当RegionServer宕机后，将RegionServer上的hlog拆分，然后分配给不同的RegionServer加载，修改.META。
  
• HLog的数据会同步到HDFS，保证数据的可靠性。
  

三、Api的使用

1. package com.xiaojie.hadoop.utils;
3. import lombok.extern.slf4j.Slf4j;
4. import org.apache.commons.lang3.StringUtils;
5. import org.apache.hadoop.conf.Configuration;
6. import org.apache.hadoop.hbase.HBaseConfiguration;
7. import org.apache.hadoop.hbase.NamespaceDescriptor;
8. import org.apache.hadoop.hbase.TableName;
9. import org.apache.hadoop.hbase.client.*;
10. import org.apache.hadoop.hbase.filter.FilterList;
11. import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
12. import org.apache.hadoop.hbase.quotas.QuotaSettings;
13. import org.apache.hadoop.hbase.quotas.QuotaType;
14. import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
15. import org.apache.hadoop.hbase.util.Bytes;
16. import org.apache.hadoop.hbase.util.Pair;
18. import java.io.IOException;
19. import java.util.List;
21. /**
22. * @author 熟透的蜗牛
23. * @version 1.0
24. * @description: hbase工具类
25. * @date 2025/1/3 9:27
26. */
27. @Slf4j
28. public class HBaseUtil {
30.     private static Connection connection;
32.     static {
33.         Configuration configuration = HBaseConfiguration.create();
34.         //设置端口号
35.         configuration.set("hbase.zookeeper.property.clientPort", "2181");
36.         //设置zk连接
37.         configuration.set("hbase.zookeeper.quorum", "hadoop1,hadoop2,hadoop3");
38.         //创建连接
39.         try {
40.             connection = ConnectionFactory.createConnection(configuration);
41.         } catch (IOException e) {
42.             throw new RuntimeException(e);
43.         }
44.     }
47.     /**
48.      * @param namespace
49.      * @description: 创建namespace
50.      * @return: boolean
51.      * @author 熟透的蜗牛
52.      * @date: 2025/1/3 11:03
53.      */
54.     public static boolean createNameSpace(String namespace) {
55.         try {
56.             Admin admin = connection.getAdmin();
57.             NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
58.             admin.createNamespace(namespaceDescriptor);
59.             log.info(">>>>>>>>>>>>创建namespace成功");
60.             return true;
61.         } catch (IOException e) {
62.             throw new RuntimeException(e);
63.         }
64.     }
66.     /**
67.      * @param namespace
68.      * @description: 删除ns
69.      * @return: boolean
70.      * @author 熟透的蜗牛
71.      * @date: 2025/1/3 11:05
72.      */
73.     public static boolean deleteNameSpace(String namespace) {
74.         try {
75.             Admin admin = connection.getAdmin();
76.             admin.deleteNamespace(namespace);
77.             log.info(">>>>>>>>>>>>删除namespace成功");
78.             return true;
79.         } catch (IOException e) {
80.             throw new RuntimeException(e);
81.         }
82.     }
84.     /**
85.      * @param tableName      表名
86.      * @param columnFamilies 列族
87.      * @description: 创建表
88.      * @return: boolean
89.      * @author 熟透的蜗牛
90.      * @date: 2025/1/3 9:35
91.      */
92.     public static boolean createTable(String tableName, List<String> columnFamilies, String nameSpace) throws IOException {
93.         Admin admin = connection.getAdmin();
94.         boolean exists = admin.tableExists(TableName.valueOf(tableName));
95.         //创建表
96.         if (!exists) {
97.             //如果namespace是空值则会使用default，作为命名空间
98.             TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(nameSpace, tableName));
99.             columnFamilies.forEach(cf -> {
100.                 ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
101.                 columnFamilyDescriptorBuilder.setMaxVersions(1);
102.                 ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
103.                 tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
104.             });
105.             admin.createTable(tableDescriptorBuilder.build());
106.             return true;
107.         } else {
108.             log.info("table exists>>>>>>>>");
109.             return false;
110.         }
111.     }
113.     /**
114.      * @param tableName 表名
115.      * @description: 删除表
116.      * @return: boolean
117.      * @author 熟透的蜗牛
118.      * @date: 2025/1/3 10:16
119.      */
120.     public static boolean deleteTable(String tableName) {
121.         try {
122.             Admin admin = connection.getAdmin();
123.             //先禁用表
124.             admin.disableTable(TableName.valueOf(tableName));
125.             //再删除
126.             admin.deleteTable(TableName.valueOf(tableName));
127.             return true;
128.         } catch (IOException e) {
129.             throw new RuntimeException(e);
130.         }
131.     }
133.     /**
134.      * @param tableName        表名
135.      * @param rowKey           rowkey
136.      * @param columnFamilyName 列族
137.      * @param qualifier        列标识
138.      * @param value            数据
139.      * @description: 插入数据
140.      * @return: boolean
141.      * @author 熟透的蜗牛
142.      * @date: 2025/1/3 16:46
143.      */
144.     public static boolean putRow(String nameSpace, String tableName, String rowKey, String columnFamilyName, String qualifier,
145.                                  String value) {
146.         Table table = null;
147.         try {
148.             table = connection.getTable(TableName.valueOf(nameSpace, tableName));
149.             Put put = new Put(Bytes.toBytes(rowKey));
150.             put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
151.             table.put(put);
152.             log.info(">>>>>>>插入数据成功");
153.             return true;
154.         } catch (IOException e) {
155.             throw new RuntimeException(e);
156.         } finally {
157.             if (table != null) {
158.                 try {
159.                     table.close();
160.                 } catch (IOException e) {
161.                     throw new RuntimeException(e);
162.                 }
163.             }
164.         }
165.     }
167.     /**
168.      * @param nameSpace
169.      * @param tableName
170.      * @param rowKey
171.      * @param columnFamilyName
172.      * @param pairList         键值对集合
173.      * @description: 插入数据
174.      * @return: boolean
175.      * @author 熟透的蜗牛
176.      * @date: 2025/1/3 17:32
177.      */
178.     public static boolean putRow(String nameSpace, String tableName, String rowKey, String columnFamilyName, List<Pair<String, String>> pairList) {
180.         Table table = null;
181.         try {
182.             table = connection.getTable(TableName.valueOf(nameSpace, tableName));
183.             Put put = new Put(Bytes.toBytes(rowKey));
184.             pairList.forEach(pair -> {
185.                 put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getFirst()), Bytes.toBytes(pair.getSecond()));
186.             });
187.             table.put(put);
188.             log.info(">>>>>>>插入数据成功");
189.             return true;
190.         } catch (IOException e) {
191.             throw new RuntimeException(e);
192.         } finally {
193.             if (table != null) {
194.                 try {
195.                     table.close();
196.                 } catch (IOException e) {
197.                     throw new RuntimeException(e);
198.                 }
199.             }
200.         }
201.     }
203.     /**
204.      * @param nameSpace
205.      * @param tableName
206.      * @param rowKey
207.      * @description:根据Rowkey查询
208.      * @return: org.apache.hadoop.hbase.client.Result
209.      * @author 熟透的蜗牛
210.      * @date: 2025/1/3 17:42
211.      */
212.     public static Result getRowByRowKey(String nameSpace, String tableName, String rowKey) {
213.         try {
214.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
215.             Result result = table.get(new Get(Bytes.toBytes(rowKey)));
216.             return result;
217.         } catch (IOException e) {
218.             throw new RuntimeException(e);
219.         }
220.     }
222.     /**
223.      * @param nameSpace
224.      * @param tableName
225.      * @description: 查询所有数据, 和范围查询
226.      * @return: org.apache.hadoop.hbase.client.ResultScanner
227.      * @author 熟透的蜗牛
228.      * @date: 2025/1/3 18:12
229.      */
230.     public static ResultScanner getAll(String nameSpace, String tableName) {
231.         try {
232.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
233.             Scan scan = new Scan();
234.             scan.setCacheBlocks(true);//设置读缓存
235.             scan.withStartRow(Bytes.toBytes("1002")); //rowkey的起始值
236.             scan.withStopRow(Bytes.toBytes("1003"));  //rowkey的结束值，返回结果不包含该值
237.             ResultScanner scanner = table.getScanner(scan);
238.             return scanner;
239.         } catch (IOException e) {
240.             throw new RuntimeException(e);
241.         }
242.     }
244.     /**
245.      * @param nameSpace
246.      * @param tableName
247.      * @param filterList
248.      * @description: 查找过滤
249.      * @return: org.apache.hadoop.hbase.client.ResultScanner
250.      * @author 熟透的蜗牛
251.      * @date: 2025/1/3 20:47
252.      */
253.     public static ResultScanner getScanner(String nameSpace, String tableName, FilterList filterList) {
254.         try {
255.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
256.             Scan scan = new Scan();
257.             scan.setFilter(filterList);
258.             ResultScanner scanner = table.getScanner(scan);
259.             return scanner;
260.         } catch (IOException e) {
261.             throw new RuntimeException(e);
262.         }
263.     }
265.     /**
266.      * @param tableName
267.      * @param rowKey
268.      * @param columnFamily 列族
269.      * @param qualifier    限定符
270.      * @description: 获取指定cell数据
271.      * @return: java.lang.String
272.      * @author 熟透的蜗牛
273.      * @date: 2025/1/3 20:59
274.      */
275.     public static String getCell(String nameSpace, String tableName, String rowKey, String columnFamily, String qualifier) {
276.         try {
277.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
278.             Get get = new Get(Bytes.toBytes(rowKey));
279.             if (!get.isCheckExistenceOnly()) {
280.                 get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
281.                 Result result = table.get(get);
282.                 byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
283.                 return Bytes.toString(resultValue);
284.             }
286.         } catch (IOException e) {
287.             throw new RuntimeException(e);
288.         }
289.         return null;
290.     }
292.     /**
293.      * @param nameSpace
294.      * @param tableName
295.      * @param rowKey
296.      * @description: 删除一行数据
297.      * @return: boolean
298.      * @author 熟透的蜗牛
299.      * @date: 2025/1/3 21:34
300.      */
301.     public static boolean deleteRow(String nameSpace, String tableName, String rowKey) {
302.         try {
303.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
304.             Delete delete = new Delete(Bytes.toBytes(rowKey));
305.             table.delete(delete);
306.         } catch (IOException e) {
307.             e.printStackTrace();
308.         }
309.         return true;
310.     }
312.     /**
313.      * @param nameSpace
314.      * @param tableName
315.      * @param rowKey
316.      * @param familyName
317.      * @param qualifier
318.      * @description: 删除指定列
319.      * @return: boolean
320.      * @author 熟透的蜗牛
321.      * @date: 2025/1/3 21:34
322.      */
323.     public static boolean deleteColumn(String nameSpace, String tableName, String rowKey, String familyName,
324.                                        String qualifier) {
325.         try {
326.             Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
327.             Delete delete = new Delete(Bytes.toBytes(rowKey));
328.             delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
329.             table.delete(delete);
330.             table.close();
331.         } catch (IOException e) {
332.             e.printStackTrace();
333.         }
334.         return true;
335.     }
338. }

复制代码

四、Phoenix

Phoenix是 HBase 的开源 SQL 中心层，它允许你使用标准 JDBC 的方式来操作 HBase 上的数据。
1、安装Phoenix

1. #解压文件
2. tar -zxf phoenix-hbase-2.6-5.2.1-bin.tar.gz
4. #移动文件
5. mv phoenix-hbase-2.6-5.2.1-bin /usr/local/
7. #复制jar到regionserver 和master的lib
8. cp phoenix-server-hbase-2.6.jar /usr/local/hbase-2.6.1/lib
10. #分发到其他服务器
11. xsync /usr/local/hbase-2.6.1/
13. #重启hbase
14. ./hbase.sh stop
15. ./hbase.sh start
17. #启动phoenix
18. /usr/local/phoenix-hbase-2.6-5.2.1-bin/bin/sqlline.py hadoop1,hadoop2,hadoop3

复制代码

2、简单使用

2.1、查看表信息

1. #查看表信息
2. !tables

复制代码

2.2、创建表

1. CREATE TABLE IF NOT EXISTS us_population (
2.       state CHAR(2) NOT NULL,
3.       city VARCHAR NOT NULL,
4.       population BIGINT
5.       CONSTRAINT my_pk PRIMARY KEY (state, city));

复制代码

2.3、插入数据

1. #插入数据
2. UPSERT INTO us_population VALUES('NY','New York',8143197);

复制代码

2.4、查看数据

1. select * from us_population;

复制代码

2.5、修改数据

1. UPSERT INTO us_population VALUES('NY','New York',999999);

复制代码

2.6、删除数据

1. DELETE FROM us_population WHERE city='New York';
3. #删除表
4. drop table us_population;

复制代码

2.7、退出

1. !quit

复制代码

2.8、导入数据

us_population.csv

1. NY,New York,8143197
2. CA,Los Angeles,3844829
3. IL,Chicago,2842518
4. TX,Houston,2016582
5. PA,Philadelphia,1463281
6. AZ,Phoenix,1461575
7. TX,San Antonio,1256509
8. CA,San Diego,1255540
9. TX,Dallas,1213825
10. CA,San Jose,912332

复制代码

us_population.sql

1. CREATE TABLE IF NOT EXISTS us_population (
2.   state CHAR(2) NOT NULL,
3.   city VARCHAR NOT NULL,
4.   population BIGINT
5.   CONSTRAINT my_pk PRIMARY KEY (state, city));

复制代码

1. #导入数据
2. ./psql.py hadoop1,hadoop2,hadoop3 us_population.sql us_population.csv
4. #查看数据
5. select * from us_population;
7. #查询数据
8. SELECT state as "State",count(city) as "City Count",sum(population) as "Population Sum"
9. FROM us_population
10. GROUP BY state
11. ORDER BY sum(population) DESC;

复制代码

3、API的使用

1. package com.xiaojie;
3. import java.sql.*;
5. /**
6. * @author 熟透的蜗牛
7. * @version 1.0
8. * @description: 测试phoenix
9. * @date 2025/1/3 23:08
10. */
11. public class PhoenixApi {
13.     public static void main(String[] args) throws ClassNotFoundException, SQLException {
14.         //1、加载驱动
15.         Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
17.         //2、创建连接
18.         String url = "jdbc:phoenix:hadoop1,hadoop2,hadoop3";
19.         Connection connect = DriverManager.getConnection(url);
20.         //3、创建查询
21.         PreparedStatement preparedStatement = connect.prepareStatement("SELECT * FROM us_population");
23.         //4、遍历结果
24.         ResultSet resultSet = preparedStatement.executeQuery();
25.         while (resultSet.next()) {
26.             System.out.println(resultSet.getString("city") + " "
27.                                + resultSet.getInt("population"));
28.         }
29.         //5、关闭资源
30.         preparedStatement.close();
31.         resultSet.close();
32.         connect.close();
33.     }
34. }

复制代码

五、整合SpringBoot+MyBatis

1. package com.xiaojie.hadoop.hbase.phoenix.dao;
3. import com.xiaojie.hadoop.hbase.phoenix.bean.USPopulation;
4. import org.apache.ibatis.annotations.*;
6. import java.util.List;
8. @Mapper
9. public interface PopulationDao {
11.     @Select("SELECT * from us_population")
12.     List<USPopulation> queryAll();
14.     @Insert("UPSERT INTO us_population VALUES( #{state}, #{city}, #{population} )")
15.     void save(USPopulation USPopulation);
17.     @Select("SELECT * FROM us_population WHERE state=#{state} AND city = #{city}")
18.     USPopulation queryByStateAndCity(String state, String city);
21.     @Delete("DELETE FROM us_population WHERE state=#{state} AND city = #{city}")
22.     void deleteByStateAndCity(String state, String city);
24. }

复制代码

1. server:
2.   port: 9999
3. spring:
4.   application:
5.     name: hadoop
6.   datasource:
7.     #zookeeper地址
8.     url: jdbc:phoenix:hadoop1,hadoop2,hadoop3
9.     driver-class-name: org.apache.phoenix.jdbc.PhoenixDriver
10.     type: com.zaxxer.hikari.HikariDataSource
11.     hikari:
12.       # 池中维护的最小空闲连接数
13.       minimum-idle: 10
14.       # 池中最大连接数，包括闲置和使用中的连接
15.       maximum-pool-size: 20
16.       # 此属性控制从池返回的连接的默认自动提交行为。默认为true
17.       auto-commit: true
18.       # 允许最长空闲时间
19.       idle-timeout: 30000
20.       # 此属性表示连接池的用户定义名称，主要显示在日志记录和JMX管理控制台中，以标识池和池配置。 默认值：自动生成
21.       pool-name: custom-hikari
22.       #此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
23.       max-lifetime: 1800000
24.       # 数据库连接超时时间,默认30秒，即30000
25.       connection-timeout: 30000
26.       # 连接测试sql 这个地方需要根据数据库方言差异而配置 例如 oracle 就应该写成  select 1 from dual
27.       connection-test-query: SELECT 1
28. wssnail:
29.   hdfs:
30.     url: hdfs://hadoop1:8020
31.     user-name: root
32. # mybatis 相关配置
33. mybatis:
34.   configuration:
35.     # 是否打印sql语句 调试的时候可以开启
36.     log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

复制代码

六、参考

https://phoenix.apache.org/Phoenix-in-15-minutes-or-less.html
完整代码：spring-boot: Springboot整合redis、消息中心件等相关代码 - Gitee.com

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。