第一章.HBase进阶

1.RegionServer架构

$02[HBase进阶] - 图1

  1. # storeFile
  2. - 保存实际数据的物理文件,StoreFile以Hfile的形式存储在HDFS 上,每个Store会有一个或多个StoreFile(Hfile),数据在每个StoreFile都是有序的
  3. # MemStore
  4. - 写缓存,由于Hfile中的数据要求是有序的,所以数据是先存储在MemStore中,排好序后,等到达刷写时机才会刷写到HFile,每次刷写都会形成一个新的HFile
  5. # WAL
  6. - 由于数据要经MemStore排序后才能刷写到HFile,但把数据保存在内存中会有很高的概率导致数据丢失,为了解决这个问题,数据会先写在一个叫做Write-Ahead logfile的文件中,然后再写入MemStore中,所以在系统出现故障的时候,数据可以通过这个日志文件重建
  7. # BlockCache
  8. - 读缓存,每次查询出的数据会缓存在BlockCache中,方便下次查询

2.写流程

$02[HBase进阶] - 图2

写流程:

  1. Client先访问zookeeper,获取hbase:meta表位于哪个Region Server
  2. 访问对应的Region Server,获取hbase:meta表根据读请求的namespace:table/rowkey,查询出目标数据位于哪个Region Server中的哪个Region中,并将该table的region信息以及meta表的位置信息缓存在客户端的meta cache,方便下次访问
  3. 与目标Region Server进行通讯
  4. 将数据顺序写入(追加)到WAL
  5. 将数据写入对应的MemStore,数据会在MemStore进行排序
  6. 向客户端发送ack
  7. 等到达MemStore的刷写时机后,将数据刷写到Hfile

3.MemStore Flush(刷写时机)

$02[HBase进阶] - 图3

  1. 1. 当某个MemStore的大小达到了hbase.hregion.memstore.flush.size(默认值128M),其所在region的所有memstore都会刷写:
  2.     当memstore的大小达到了  hbase.hregion.memstore.flush.size(默认值128M)  *hbase.hregion.memstore.block.multiplier(默认值4)时,会阻止继续往memstore写数据
  4. 2. 当region server中的memstore的总大小达到
  6. java_heapsize
  7. *hbase.regionserver.global.memstore.size（默认值0.4）
  8. *hbase.regionserver.global.memstore.size.lower.limit（默认值0.95)
  10. region会按照其所有的memstore大小顺序(由大到小)依次进行刷写,直到region server中所有的memstore的总大小减小到上述值以下
  12. 当region server中memstore的总大小达到
  13. java_heapsize
  14. *hbase.regionserver.global.memstore.size(默认值0.4)
  16. 时,会阻止继续往所有的memstore写数据
  18. 3. 到达自动刷写的时间,也会触发memstore flush,自动刷新的时间间隔由该属性进行配置hbase.regionserver.optionalcacheflushinterval(默认1小时)
  19. 4. 当WAL文件的数量超过hbase.regionserver.max.logs,region会按照时间顺序依次进行刷写,直到WAL文件数量减少到hbase.regionserver.max.logs以下(该属性名已经废弃,现无需手动设置,最大值为32)

4.读流程

  1. 整体流程

$02[HBase进阶] - 图4

  1. Merge细节

$02[HBase进阶] - 图5

读流程:

  1. Client先访问zookeeper,获取hbase:meta表位于哪个Region Server
  2. 访问对应的Region Server, 获取hbase:meta表,根据读请求的namespace:table/rowkey,查询出目标数据位于哪个Region中,并将该table的region信息以及meta表的位置信息缓存在客户端的meta cache,方便下次访问
  3. 与目标Region Server进行通讯
  4. 分别在MemStore和Store File(Hfile)中查询目标数据,并将查到的所有数据进行合并,此处所有数据是指同一条数据的不同版本(time stamp)或者不同的类型(Put/Delete)
  5. 将查询到的新的数据块(Block,Hfile数据存储单元,默认大小为64kb)缓存到Block Cache
  6. 将合并后的最终结果返回给客户端

5.StoreFile Compaction

$02[HBase进阶] - 图6

  1. - 由于memstore每次刷写都会生成一个新的Hfile,且同一个字段的不同版本(timestamp)和不同类型(Put/Delete)有可能会分布在不同的hfile中,因此查询时需要遍历所有的Hfile,为了减少Hfile的个数,以及清理掉过期和删除的数据,会进行StoreFile Compaction
  2. - Compaction分为两种,分别是Minor Compaction 和 Major Compaction,Minor Compaction会将临近的若干个较小的Hfile合并成一个较大的Hfile,并清理掉部分过期和删除的数据,Major Compaction 会将一个Store下的所有Hfile合并成一个大的Hfile,并且会清理掉所有过期和删除的数据

6.Region Split

  1. - 默认情况下,每个Table起初只有一个Region,随着数据的不断写入,Region会自动进行拆分,刚拆分时,两个子Region都位于当前的Region Server,但随着负载均衡的考虑,HMaster有可能会将某个Region转移给其他的Region Server
  3. - Region Split时机:
  4. 1. 当一个Region中的某个Store下所有的StoreFile的总大小超过hbase.hregion.max.filesize,该Region就会进行拆分(0.94版本之前)
  5. 2. 当一个region中的某个Store下所有StoreFile的总大小超过Min(initialSize*R^3,hbase.hregion.max.filesize),该region就会进行拆分,其中initialSize的默认值为2*
  6. hbase.hregion.memstore.flush.size,R为当前Region Server中属于该Table的Region个数(0.94版本之后)
  8. 具体的切分策略为:
  9. 第一次 split :1^3*256 =256MB
  10. 第二次split：2^3 * 256 = 2048MB 
  11. 第三次split：3^3 * 256 = 6912MB 
  12. 第四次split：4^3 * 256 = 16384MB > 10GB，因此取较小的值10GB 
  13. 后面每次split的size都是10GB了
  15. 3. HBase2.0引入了新的split策略,如果当前RegionServer上该表只有一个Region,按照2* hbase.hregion.memstore.flush.size分裂,否则按照hbase.hregion.max.filesize分裂

$02[HBase进阶] - 图7

第二章.HBase API

1.环境准备

新建一个maven项目,并添加依赖

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

2.DDL

  1. package com.atguigu.hbaseapi;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.hbase.HBaseConfiguration;
  5. import org.apache.hadoop.hbase.NamespaceDescriptor;
  6. import org.apache.hadoop.hbase.TableName;
  7. import org.apache.hadoop.hbase.client.*;
  8. import org.apache.hadoop.hbase.util.Bytes;
  10. import java.io.IOException;
  11. import java.util.List;
  13. public class HBaseDDL {
  14.     public static Connection connection;
  16.     static{
  17.         try{
  18.             //创建配置对象
  19.             Configuration configuration = HBaseConfiguration.create();
  20.             configuration.set("hbase.zookeeper.quorum","hadoop102:2181");
  21.             //获取连接对象
  22.             connection = ConnectionFactory.createConnection(configuration);
  23.         }catch(IOException e){
  24.             e.printStackTrace();
  25.         }
  26.     }
  28.     /** 
  29.     * @Description: 创建命名空间
  30.     * @Param: [nameSpace] 
  31.     * @return: void 
  32.     * @Author: jcsune
  33.     * @Date: 2021/8/2 
  34.     */ 
  35.     public static void createNameSpace(String nameSpace) throws IOException{
  36.         //获取admin
  37.         Admin admin = connection.getAdmin();
  38.         //构建NameSpaceDescriptor
  39.         NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(nameSpace);
  40.         NamespaceDescriptor namespaceDescriptor = builder.build();
  41.         //利用admin创建命名空间
  42.         admin.createNamespace(namespaceDescriptor);
  43.         System.out.println("创建命名空间" + nameSpace + "成功");
  44.         //关闭资源
  45.         admin.close();
  46.     }
  47.     /**
  48.     * @Description: 列出当前所有的命名空间
  49.     * @Param: []
  50.     * @return: void
  51.     * @Author: jcsune
  52.     * @Date: 2021/8/2
  53.     */
  54.     public static void listNameSpace()throws IOException{
  55.         //获取admin
  56.         Admin admin = connection.getAdmin();
  57.         //获取所有命名空间的信息
  58.         NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
  59.         System.out.println("所有的命名空间:");
  60.         for (NamespaceDescriptor namespaceDescriptor : namespaceDescriptors) {
  61.             System.out.println(namespaceDescriptor.getName());
  62.         }
  63.         //关闭命名空间
  64.         admin.close();
  67.     }
  68.     /**
  69.     * @Description: 删除命名空间
  70.     * @Param: [nameSpace]
  71.     * @return: void
  72.     * @Author: jcsune
  73.     * @Date: 2021/8/2
  74.     */
  75.     public static void dropNameSpace(String nameSpace) throws IOException{
  76.         //获取admin
  77.         Admin admin = connection.getAdmin();
  78.         //删除命名空间
  79.         admin.deleteNamespace(nameSpace);
  80.         System.out.println("删除命名空间" + nameSpace + "成功");
  81.         //关闭资源
  82.         admin.close();
  83.     }
  85.     /**
  86.     * @Description: 创建表
  87.     * @Param: [tableName, cf]
  88.     * @return: void
  89.     * @Author: jcsune
  90.     * @Date: 2021/8/2
  91.     */
  92.     public static void createTable(String tableName,String cf) throws IOException{
  93.         //获取admin
  94.         Admin admin = connection.getAdmin();
  95.         //判断
  96.         if(existsTable(tableName)){
  97.             System.out.println(tableName+ "表已将存在了");
  98.             return;
  99.         }
  100.         //构建表描述者构造器
  101.         TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
  102.         //创建一个列族描述者构造器
  103.         ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
  104.         ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
  105.         //向表描述者构造器设置列族信息
  106.         tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
  107.         TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
  108.         //创建表
  109.         admin.createTable(tableDescriptor);
  110.         System.out.println(tableName+"表创建成功");
  111.         //关闭资源
  112.         admin.close();
  115.     }
  116.     /**
  117.     * @Description: 删除表
  118.     * @Param: [tableName]
  119.     * @return: void
  120.     * @Author: jcsune
  121.     * @Date: 2021/8/2
  122.     */
  123.     public static void dropTable(String tableName) throws IOException{
  124.         //获取admin
  125.         Admin admin = connection.getAdmin();
  126.         //判断
  127.         if (!existsTable(tableName)){
  128.             System.out.println(tableName + "表不存在");
  129.             admin.close();
  130.             return;
  131.         }
  132.         //删除表
  133.         admin.deleteTable(TableName.valueOf(tableName));
  134.         System.out.println(tableName+"表删除成功");
  135.         //关闭资源
  136.         admin.close();
  137.     }
  139.     /**
  140.     * @Description: 列出所有表
  141.     * @Param: []
  142.     * @return: void
  143.     * @Author: jcsune
  144.     * @Date: 2021/8/2
  145.     */
  146.     public static void listTable() throws IOException{
  147.         //获取admin
  148.         Admin admin = connection.getAdmin();
  149.         //列出表
  150.         List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
  151.         //遍历
  152.         System.out.println("所有的表:");
  153.         for (TableDescriptor tableDescriptor : tableDescriptors) {
  154.             System.out.println(tableDescriptor.getTableName());
  155.         }
  156.         //关闭资源
  157.         admin.close();
  158.     }
  162.     /**
  163.     * @Description: 判断表是否存在
  164.     * @Param: [tableName]
  165.     * @return: boolean
  166.     * @Author: jcsune
  167.     * @Date: 2021/8/2
  168.     */
  169.     private static boolean existsTable(String tableName) throws IOException{
  170.         //获取admin
  171.         Admin admin = connection.getAdmin();
  172.         //判断
  173.         boolean exists = admin.tableExists(TableName.valueOf(tableName));
  174.         return exists;
  175.     }
  178.     public static void main(String[] args) {
  179.         try {
  180.             //HBaseDDL.createNameSpace("xiyouji");
  181.             //listNameSpace();
  182.             //dropNameSpace("xiyouji");
  183.             listTable();
  184.             createTable("xiyouji","info");
  185.             dropTable("xiyouji");
  187.         } catch (IOException e) {
  188.             e.printStackTrace();
  189.         }
  190.     }
  191. }

3.DML

  1. package com.atguigu.hbase.API.demo.DML;
  3. import com.atguigu.hbase.API.demo.DDL.HBaseDDL;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.hbase.Cell;
  6. import org.apache.hadoop.hbase.CellUtil;
  7. import org.apache.hadoop.hbase.HBaseConfiguration;
  8. import org.apache.hadoop.hbase.TableName;
  9. import org.apache.hadoop.hbase.client.*;
  10. import org.apache.hadoop.hbase.util.Bytes;
  12. import java.awt.event.WindowFocusListener;
  13. import java.io.IOException;
  15. /**
  16.  * @author leon
  17.  * @ClassName HBaseDML.java
  18.  * @createTime 2021年07月31日 11:11:00
  19.  */
  20. public class HBaseDML {
  22.     public static Connection connection;
  24.     static {
  25.         try {
  26.             // 1.创建配置对象
  27.             Configuration configuration = HBaseConfiguration.create();
  28.             configuration.set("hbase.zookeeper.quorum", "hadoop102:2181");
  29.             // 2. 获取连接对象
  30.             connection = ConnectionFactory.createConnection(configuration);
  31.         } catch (IOException e) {
  32.             e.printStackTrace();
  33.         }
  34.     }
  36.     /**
  37.      * 插入/修改
  38.      * @param tableName
  39.      * @param rowkey
  40.      * @param cf
  41.      * @param cn
  42.      * @param value
  43.      */
  44.     public static void putData(String tableName, String rowkey, String cf, String cn, String value) throws IOException {
  45.         // 1. 获取Table对象
  46.         Table table = connection.getTable(TableName.valueOf(tableName));
  47.         // 2. 构建put对象
  48.         Put put = new Put(Bytes.toBytes(rowkey));
  49.         // 3. 设置列族、列名属性
  50.         put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cn),Bytes.toBytes(value));
  51.         // 4. 执行插入
  52.         table.put(put);
  53.         System.out.println("数据插入成功");
  54.         // 5. 关闭资源
  55.         table.close();
  56.     }
  58.     /**
  59.      * get
  60.      * @param tableName
  61.      * @param rowkey
  62.      * @param cf
  63.      * @param cn
  64.      */
  65.     public static void getData(String tableName, String rowkey, String cf, String cn) throws IOException {
  66.         // 1. 获取Table对象
  67.         Table table = connection.getTable(TableName.valueOf(tableName));
  68.         // 2. 构建get对象
  69.         Get get = new Get(Bytes.toBytes(rowkey));
  70.         // 3. 设置列族、列名属性
  71.         if(cf == null && cn != null){
  72.             System.out.println("输入有误");
  73.             table.close();
  74.             return;
  75.         }
  76.         if(cf != null && cn == null){
  77.             // 查询某列族
  78.             get.addFamily(Bytes.toBytes(cf));
  79.         }
  80.         if(cf != null && cn != null){
  81.             // 查询cell
  82.             get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn));
  83.         }
  85.         // 4. 执行插入
  86.         Result result = table.get(get);
  87.         // 5. 解析result
  88.         Cell[] cells = result.rawCells();
  89.         // 6. 遍历
  90.         System.out.println("数据是：");
  91.         for (Cell cell : cells) {
  92.             String rk = Bytes.toString(cell.getRowArray());
  93.             String rowkey1 = Bytes.toString(CellUtil.cloneRow(cell));
  94.             String cFamily = Bytes.toString(CellUtil.cloneFamily(cell));
  95.             String cname = Bytes.toString(CellUtil.cloneQualifier(cell));
  96.             String value = Bytes.toString(CellUtil.cloneValue(cell));
  97.             System.out.println("rk:"+rk);
  98.             System.out.println("rowkey1:"+rowkey1+", cf:cn"+cFamily+":"+cname+", value:"+value);
  99.         }
  100.         // 5. 关闭资源
  101.         table.close();
  102.     }
  104.     /**
  105.      * 扫描表
  106.      * @param tableName
  107.      */
  108.     public static void scanData(String tableName) throws IOException {
  109.         // 1. 获取Table对象
  110.         Table table = connection.getTable(TableName.valueOf(tableName));
  111.         // 2. 创建scan
  112.         Scan scan = new Scan();
  113.         // 3. 扫描
  114.         ResultScanner scanner = table.getScanner(scan);
  115.         System.out.println("数据是：");
  116.         for (Result result : scanner) {
  117.             Cell[] cells = result.rawCells();
  118.             for (Cell cell : cells) {
  119.                 String rowkey1 = Bytes.toString(CellUtil.cloneRow(cell));
  120.                 String cFamily = Bytes.toString(CellUtil.cloneFamily(cell));
  121.                 String cname = Bytes.toString(CellUtil.cloneQualifier(cell));
  122.                 String value = Bytes.toString(CellUtil.cloneValue(cell));
  123.                 System.out.println("rowkey1:"+rowkey1+", cf:cn"+cFamily+":"+cname+", value:"+value);
  124.             }
  125.         }
  126.         // 5. 关闭资源
  127.         table.close();
  128.     }
  130.     /**
  131.      * 删除数据
  132.      */
  133.     public static void dropData(String tableName, String rowKey) throws IOException {
  134.         // 1. 获取table
  135.         Table table = connection.getTable(TableName.valueOf(tableName));
  136.         // 2. 创建delete对象
  137.         Delete delete = new Delete(Bytes.toBytes(rowKey));
  138.         // 3. 删除一行数据
  139.         table.delete(delete);
  140.         System.out.println("删除数据成功");
  141.         // 4. 关闭资源
  142.         table.close();
  143.     }
  145.     public static void main(String[] args) {
  146.         try {
  147. //            HBaseDDL.createTable("xixi", "info");
  148. //            putData("xixi", "1001", "info","name", "张三");
  149. //            putData("xixi", "1001", "info","age", "23");
  150. //            putData("xixi", "1001", "info","sex", "难");
  151. //            System.out.println("=====查询一行======");
  152. //            getData("xixi", "1001", null, null);
  153. //            System.out.println("=====查询一列族======");
  154. //            getData("xixi", "1001", "info", null);
  155. //            System.out.println("=====查cell======");
  156. //            getData("xixi", "1001", "info", "sex");
  158. //            scanData("xixi");
  159. //            dropData("xixi", "1001");
  160. //            scanData("xixi");
  161.             getData("STUDENT", "1001", null, "name");
  162.         } catch (IOException e) {
  163.             e.printStackTrace();
  164.         }
  165.     }
  167. }

第三章.HBase优化

1.预分区

每一个region维护着startRow与endRow,如果加入的数据符合某个region维护的rowKey范围,则该数据交给这个region维护,那么依照这个原则,我们可以将数据将要投放的分区提前大致的规划好,以提高HBase性能

  1. 手动设定预分区
  1. create 'xixi','info',SPILTS =>['1000','2000','3000','4000']
  1. 生成16进制序列预分区
  1. create 'xixi','info',{NUMREGIONS => 15, SPLITALGO => 'HexStringSplit'}
  1. 按照文件设置的规则预分区

创建splits.txt,文件内容如下

  1. aaaa
  2. bbbb
  3. cccc
  4. dddd

然后执行

  1. create 'xixi','info',SPLITS FILE  => 'splits.txt'
  1. 使用JavaAPI创建预分区
  1. //自定义算法，产生一系列Hash散列值存储在二维数组中
  2. byte[][] splitKeys = 某个散列值函数
  3. //创建HbaseAdmin实例
  4. HBaseAdmin hAdmin = new HBaseAdmin(HbaseConfiguration.create());
  5. //创建HTableDescriptor实例
  6. HTableDescriptor tableDesc = new HTableDescriptor(tableName);
  7. //通过HTableDescriptor实例和散列值二维数组创建带有预分区的Hbase表
  8. hAdmin.createTable(tableDesc, splitKeys);

2.RowKey设计

一条数据的唯一标识就是rowkey,那么这条数据存储于哪个分区,取决于rowkey处于哪一个预分区的区间内,设计rowkey的主要目的,就是让数据均匀的分布于所有的region中,在一定程度上防止数据倾斜

3.内存优化

HBase操作过程中需要大量的内存开销,毕竟Table是可以缓存在内存中的,但是不建议分配非常大的堆内存,因为GC过程持续太久会导致RegionServer处于长期不可用状态,一般16G~36G内存就可以了,如果因为框架占用内存过高导致系统内存不足,框架一样会被系统服务拖死

4.基础优化

  1. zookeeper会话超时时间(hbase-site.xml)
  1. # 属性：zookeeper.session.timeout
  2. # 解释：默认值为90000毫秒（90s）。当某个RegionServer挂掉，90s之后Master才能察觉到。可适当减小此值，以加快Master响应，可调整至60000毫秒。
  1. 设置RPC监听数量(hbase-site.xml)
  1. # 属性：hbase.regionserver.handler.count
  2. # 解释：默认值为30，用于指定RPC监听的数量，可以根据客户端的请求数进行调整，读写请求较多时，增加此值
  1. 手动控制Major Compaction(hbase-site.xml)
  1. # 属性：hbase.hregion.majorcompaction
  2. # 解释：默认值：604800000秒（7天）， Major Compaction的周期，若关闭自动Major Compaction，可将其设为0
  1. 优化HStore文件大小(hbase-site.xml)
  1. # 属性：hbase.hregion.max.filesize
  2. # 解释：默认值10737418240（10GB），如果需要运行HBase的MR任务，可以减小此值，因为一个region对应一个map任务，如果单个region过大，会导致map任务执行时间过长。该值的意思就是，如果HFile的大小达到这个数值，则这个region会被切分为两个Hfile。
  1. 优化Hbase客户端缓存(hbase-site.xml)
  1. # 属性：hbase.client.write.buffer
  2. # 解释：默认值2097152bytes（2M）用于指定HBase客户端缓存，增大该值可以减少RPC调用次数，但是会消耗更多内存，反之则反之。一般我们需要设定一定的缓存大小，以达到减少RPC次数的目的。
  1. 指定scan.next扫描Hbase所获取的行数(hbase-site.xml)
  1. # 属性：hbase.client.scanner.caching
  2. # 解释：用于指定scan.next方法获取的默认行数，值越大，消耗内存越大。
  1. BlockCache占用RegionServer堆内存的比例(hbase-site.xml)
  1. # 属性：hfile.block.cache.size
  2. # 解释：默认0.4，读请求比较多的情况下，可适当调大
  1. MemStore占用RegionServer堆内存的比例
  1. # 属性：hbase.regionserver.global.memstore.size
  2. # 解释：默认0.4，写请求较多的情况下，可适当调大