大数据技术之HBase3

简介: 大数据技术之HBase3

3.4.4 、带过滤扫描

 /**
 * 带过滤的扫描
 *
 * @param namespace 命名空间
 * @param tableName 表格名称
 * @param startRow 开始 row
 * @param stopRow 结束 row
 * @param columnFamily 列族名称
 * @param columnName 列名
 * @param value value 值
 * @throws IOException
 */
 public static void filterScan(String namespace, String tableName, String startRow, String stopRow, String columnFamily, String columnName, String value) throws IOException {
     // 1. 获取 table
     Table table = connection.getTable(TableName.valueOf(namespace, tableName));
     // 2. 创建 scan 对象
     Scan scan = new Scan();
     // 如果此时直接调用 会直接扫描整张表
     // 添加参数 来控制扫描的数据
     // 默认包含
     scan.withStartRow(Bytes.toBytes(startRow));
     // 默认不包含
     scan.withStopRow(Bytes.toBytes(stopRow));
     // 可以添加多个过滤
     FilterList filterList = new FilterList();
     // 创建过滤器
     // (1) 结果只保留当前列的数据
     ColumnValueFilter columnValueFilter = new ColumnValueFilter(
     // 列族名称
     Bytes.toBytes(columnFamily),
     // 列名
     Bytes.toBytes(columnName),
     // 比较关系
     CompareOperator.EQUAL,
     // 值
     Bytes.toBytes(value)
     );
     // (2) 结果保留整行数据
     // 结果同时会保留没有当前列的数据
     SingleColumnValueFilter singleColumnValueFilter = new 
    SingleColumnValueFilter(
     // 列族名称
     Bytes.toBytes(columnFamily),
     // 列名
     Bytes.toBytes(columnName),
     // 比较关系
     CompareOperator.EQUAL,
     // 值
     Bytes.toBytes(value)
     );
     // 本身可以添加多个过滤器
     filterList.addFilter(singleColumnValueFilter);
     // 添加过滤
     scan.setFilter(filterList);
     try {
         // 读取多行数据 获得 scanner
         ResultScanner scanner = table.getScanner(scan);
         // result 来记录一行数据 cell 数组
         // ResultScanner 来记录多行数据 result 的数组
         for (Result result : scanner) {
             Cell[] cells = result.rawCells();
             for (Cell cell : cells) {
             System.out.print(new 
            String(CellUtil.cloneRow(cell)) + "-" + new 
            String(CellUtil.cloneFamily(cell)) + "-" + new 
            String(CellUtil.cloneQualifier(cell)) + "-" + new 
            String(CellUtil.cloneValue(cell)) + "\t");
             }
         System.out.println();
         }
     } catch (IOException e) {
         e.printStackTrace();
     }
     // 3. 关闭 table
     table.close();
 }


骚戴理解:ColumnValueFilter和SingleColumnValueFilter的区别就是ColumnValueFilter只会查询到指定的列的cell,而SingleColumnValueFilter会把符合条件的所有列的cell查出来


使用ColumnValueFilter查询name列为张三结果如下所示

使用SingleColumnValueFilter查询name列为张三结果如下所示

这样的就很好理解了,但是这里有个1001是为什么呢?这是因为SingleColumnValueFilter除了查询符合条件的所有列,还会把没有这个列的那一行给显示出来,也就是rowKey为1001的这一行里面没有列名为name的,所以才会显示出来,由于只有一个age列,所以就只显示了一个1001-info-age-18


3.4.5、 删除数据

/**
* 删除 column 数据
*
* @param nameSpace
* @param tableName
* @param rowKey
* @param family
* @param column
* @throws IOException
*/
public static void deleteColumn(String nameSpace, String tableName, 
String rowKey, String family, String column) throws IOException {
     // 1.获取 table
     Table table = connection.getTable(TableName.valueOf(nameSpace, tableName));
     // 2.创建 Delete 对象
     Delete delete = new Delete(Bytes.toBytes(rowKey));
     // 3.添加删除信息
     // 3.1 删除单个版本
    delete.addColumn(Bytes.toBytes(family),Bytes.toBytes(column));
     // 3.2 删除所有版本
     delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(column));
     // 3.3 删除列族
    // delete.addFamily(Bytes.toBytes(family));
     // 3.删除数据
     table.delete(delete);
     // 5.关闭资源
     table.close();
    }
public static void main(String[] args) throws IOException {
    // putCell("bigdata","student","1002","info","name","lisi");
    // String cell = getCell("bigdata", "student", "1001", "info", 
    "name");
    // System.out.println(cell);
    // List<String> strings = scanRows("bigdata", "student", 
    "1001", "2000");
    // for (String string : strings) {
    // System.out.println(string);
     deleteColumn("bigdata", "student", "1001", "info", "name");
}


骚戴理解:删除有三种类型如下所示

// 3.1 删除单个版本
delete.addColumn(Bytes.toBytes(family),Bytes.toBytes(column));
// 3.2 删除所有版本
delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(column));
// 3.3 删除列族
 delete.addFamily(Bytes.toBytes(family));


4 HBase 进阶

4.1、Master 架构

Meta 表格介绍:(警告:不要去改这个表)


全称hbase:meta,只是在list 命令中被过滤掉了,本质上和HBase 的其他表格一样。


RowKey:


([table],[region start key],[region id]) 即表名、region 起始位置、regionID。


列:


regioninfo 为region 信息,存储一个HRegionInfo 对象。


server 当前region 所处的RegionServer 信息,包含端口号。


serverstartcode 当前region 被分到RegionServer 的起始时间。


如果一个表处于切分的过程中,即 region 切分,还会多出两列 info:splitA 和info:splitB,存储值也是HRegionInfo 对象,拆分结束后,删除这两列。


注意:在客户端对元数据进行操作的时候才会连接master,如果对数据进行读写,直接连接zookeeper 读取目录/hbase/meta-region-server 节点信息,会记录meta 表格的位置。直接读取即可,不需要访问master,这样可以减轻master 的压力,相当于master 专注meta 表的写操作,客户端可直接读取meta 表。


在HBase 的2.3 版本更新了一种新模式:Master Registry。客户端可以访问master 来读取meta 表信息。加大了master 的压力,减轻了zookeeper 的压力。


骚戴理解:预写日志管理器的作用就是在master做操作之前先记录在预写日志管理器里面,防止master做了一半突然挂掉了导致操作丢失,这样即使master挂掉了也可以去让备胎上位然后读取预写日志管理器继续操作


4.2、RegionServer 架构

MemStory


写缓存,由于HFile 中的数据要求是有序的,所以数据是先存储在MemStory 中,排好序后,等到达刷写时机才会刷写到HFile,每次刷写都会形成一个新的HFile,写入到对应的文件夹store 中。每个region里的每个store对应一个MemStory


WAL


由于数据要经MemStory 排序后才能刷写到HFile,但把数据保存在内存中会有很高的概率导致数据丢失,为了解决这个问题,数据会先写在一个叫做Write-Ahead logfile 的文件中,然后再写入MemStory 中。所以在系统出现故障的时候,数据可以通过这个日志文件重建。


BlockCache


读缓存,每次查询出的数据会缓存在BlockCache 中,方便下次查询。


4.3、写流程

写流程:

写流程顺序正如API 编写顺序,首先创建HBase 的重量级连接


首先访问zookeeper,获取hbase:meta 表位于哪个Region Server;


访问对应的Region Server,获取hbase:meta 表,将其缓存到连接中,作为连接属性MetaCache,由于Meta 表格具有一定的数据量,导致了创建连接比较慢;


之后使用创建的连接获取Table,这是一个轻量级的连接,只有在第一次创建的时候会检查表格是否存在访问RegionServer,之后在获取Table 时不会访问RegionServer;


调用Table 的put 方法写入数据,此时还需要解析RowKey,对照缓存的MetaCache,查看具体写入的位置有哪个RegionServer;


将数据顺序写入(追加)到WAL,此处写入是直接落盘的,并设置专门的线程控


制WAL 预写日志的滚动(类似Flume);


根据写入命令的RowKey 和ColumnFamily 查看具体写入到哪个MemStory,并且在MemStory 中排序;


向客户端发送ack;


等达到MemStore 的刷写时机后,将数据刷写到对应的story 中。


4.4、MemStore Flush

MemStore 刷写由多个线程控制,条件互相独立:

主要的刷写规则是控制刷写文件的大小,在每一个刷写线程中都会进行监控


(1)当某个 memstroe 的大小达到了 hbase.hregion.memstore.flush.size(默认值 128M),


其所在 region 的所有 memstore 都会刷写。


当 memstore 的大小达到了

hbase.hregion.memstore.flush.size(默认值 128M)* hbase.hregion.memstore.block.multiplier(默认值 4)

时,会刷写同时阻止继续往该 memstore 写数据(由于线程监控是周期性的,所有有可能面


对数据洪峰,尽管可能性比较小)


(2)由 HRegionServer 中的属性 MemStoreFlusher 内部线程 FlushHandler 控制。标准为


LOWER_MARK(低水位线)和 HIGH_MARK(高水位线),意义在于避免写缓存使用过多的内


存造成 OOM


当 region server 中 memstore 的总大小达到低水位线


java_heapsize*hbase.regionserver.global.memstore.size(默认值 0.4)*hbase.regionserver.global.memstore.size.lower.limit(默认值 0.95)

region 会按照其所有 memstore 的大小顺序(由大到小)依次进行刷写。直到 region server


中所有 memstore 的总大小减小到上述值以下。


当 region server 中 memstore 的总大小达到高水位线

java_heapsize*hbase.regionserver.global.memstore.size(默认值 0.4)

时,会同时阻止继续往所有的 memstore 写数据。


(3)为了避免数据过长时间处于内存之中,到达自动刷写的时间,也会触发 memstore flush。由 HRegionServer 的属性 PeriodicMemStoreFlusher 控制进行,由于重要性比较低,5min


才会执行一次。


自动刷新的时间间隔由该属性进行配置 hbase.regionserver.optionalcacheflushinterval(默认


1 小时)。


(4)当 WAL 文件的数量超过 hbase.regionserver.max.logs,region 会按照时间顺序依次


进行刷写,直到 WAL 文件数量减小到 hbase.regionserver.max.log 以下(该属性名已经废弃,


现无需手动设置,最大值为 32)。


4.5、 读流程

4.5.1 、HFile 结构

在了解读流程之前,需要先知道读取的数据是什么样子的。


HFile 是存储在 HDFS 上面每一个 store 文件夹下实际存储数据的文件。里面存储多种内容。包括数据本身(keyValue 键值对)、元数据记录、文件信息、数据索引、元数据索引和一个固定长度的尾部信息(记录文件的修改情况)。


键值对按照块大小(默认 64K)保存在文件中,数据索引按照块创建,块越多,索引越大。每一个 HFile 还会维护一个布隆过滤器(就像是一个很大的地图,文件中每有一种 key,就在对应的位置标记,读取时可以大致判断要 get 的 key 是否存在 HFile 中)。


KeyValue 内容如下:

rowlength -----------→ key 的长度

row -----------------→ key 的值

columnfamilylength --→ 列族长度

columnfamily --------→ 列族

columnqualifier -----→ 列名

timestamp -----------→ 时间戳(默认系统时间)

keytype -------------→ Put

由于 HFile 存储经过序列化,所以无法直接查看。可以通过 HBase 提供的命令来查看存


储在 HDFS 上面的 HFile 元数据内容。


[atguigu@hadoop102 hbase]$ bin/hbase hfile -m -f /hbase/data/命名
空间/表名/regionID/列族/HFile 名

4.5.2 、读流程

创建连接同写流程。


(1)创建 Table 对象发送 get 请求。


(2)优先访问 Block Cache,查找是否之前读取过,并且可以读取 HFile 的索引信息和


布隆过滤器。


(3)不管读缓存中是否已经有数据了(可能已经过期了),都需要再次读取写缓存和


store 中的文件。


(4)最终将所有读取到的数据合并版本,按照 get 的要求返回即可。


4.5.3 、合并读取数据优化

每次读取数据都需要读取三个位置,最后进行版本的合并。效率会非常低,所有系统需要对此优化。


(1)HFile 带有索引文件,读取对应 RowKey 数据会比较快。


(2)Block Cache 会缓存之前读取的内容和元数据信息,如果 HFile 没有发生变化(记录在 HFile 尾信息中),则不需要再次读取。


(3)使用布隆过滤器能够快速过滤当前 HFile 不存在需要读取的 RowKey,从而避免读取文件。(布隆过滤器使用 HASH 算法,不是绝对准确的,出错会造成多扫描一个文件,对读取数据结果没有影响)


4.6、StoreFile Compaction

由于memstore 每次刷写都会生成一个新的HFile,文件过多读取不方便,所以会进行文件的合并,清理掉过期和删除的数据,会进行StoreFile Compaction。


Compaction 分为两种,分别是Minor Compaction 和Major Compaction。Minor Compaction 会将临近的若干个较小的HFile 合并成一个较大的HFile,并清理掉部分过期和删除的数据,有系统使用一组参数自动控制,Major Compaction 会将一个Store 下的所有的HFile 合并成一个大HFile,并且会清理掉所有过期和删除的数据,由参数hbase.hregion.majorcompaction 控制,默认7 天。



Minor Compaction 控制机制:


参与到小合并的文件需要通过参数计算得到,有效的参数有 5 个


(1)hbase.hstore.compaction.ratio(默认 1.2F)合并文件选择算法中使用的比率。


(2)hbase.hstore.compaction.min(默认 3) 为 Minor Compaction 的最少文件个数。


(3)hbase.hstore.compaction.max(默认 10) 为 Minor Compaction 最大文件个数。


(4)hbase.hstore.compaction.min.size(默认 128M)为单个 Hfile 文件大小最小值,小于这


个数会被合并。


(5)hbase.hstore.compaction.max.size(默认 Long.MAX_VALUE)为单个 Hfile 文件大小最大


值,高于这个数不会被合并。


小合并机制为拉取整个 store 中的所有文件,做成一个集合。之后按照从旧到新的顺序遍历。


判断条件为:


① 过小合并,过大不合并


② 文件大小/ hbase.hstore.compaction.ratio < (剩余文件大小和) 则参与压缩。所有把比值设


置过大,如 10 会最终合并为 1 个特别大的文件,相反设置为 0.4,会最终产生 4 个 storeFile。


不建议修改默认值


③ 满足压缩条件的文件个数达不到个数要求(3 <= count <= 10)则不压缩。


4.7 、Region Split

Region 切分分为两种,创建表格时候的预分区即自定义分区,同时系统默认还会启动一

个切分规则,避免单个 Region 中的数据量太大。


4.7.1 、预分区(自定义分区)

每一个 region 维护着 startRow 与 endRowKey,如果加入的数据符合某个 region 维护的

rowKey 范围,则该数据交给这个 region 维护。那么依照这个原则,我们可以将数据所要投

放的分区提前大致的规划好,以提高 HBase 性能。

1)手动设定预分区


create 'staff1','info', SPLITS => ['1000','2000','3000','4000']


2)生成 16 进制序列预分区


create 'staff2','info',{NUMREGIONS => 15, SPLITALGO => 'HexStringSplit'}


3)按照文件中设置的规则预分区


(1)创建 splits.txt 文件内容如下:


aaaa
bbbb
cccc
dddd

(2)然后执行:

create 'staff3', 'info',SPLITS_FILE => 'splits.txt'

4)使用 JavaAPI 创建预分区

package com.atguigu.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
public class HBaseConnect {
 public static void main(String[] args) throws IOException {
     // 1.获取配置类
     Configuration conf = HBaseConfiguration.create();
     // 2.给配置类添加配置
    conf.set("hbase.zookeeper.quorum","hadoop102,hadoop103,hadoop104" );
     // 3.获取连接
     Connection connection = ConnectionFactory.createConnection(conf);
     // 3.获取 admin
     Admin admin = connection.getAdmin();
     // 5.获取 descriptor 的 builder
     TableDescriptorBuilder builder = 
    TableDescriptorBuilder.newBuilder(TableName.valueOf("bigdata", 
    "staff4"));
     // 6. 添加列族
    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info")).build());
     // 7.创建对应的切分
     byte[][] splits = new byte[3][];
     splits[0] = Bytes.toBytes("aaa");
     splits[1] = Bytes.toBytes("bbb");
     splits[2] = Bytes.toBytes("ccc");
     // 8.创建表
     admin.createTable(builder.build(),splits);
     // 9.关闭资源
     admin.close();
     connection.close();
     }
}

4.7.2、系统拆分

Region 的拆分是由 HRegionServer 完成的,在操作之前需要通过 ZK 汇报 master,修改对应的 Meta 表信息添加两列 info:splitA 和 info:splitB 信息。之后需要操作 HDFS 上面对应的文件,按照拆分后的 Region 范围进行标记区分,实际操作为创建文件引用,不会挪动数据。刚完成拆分的时候,两个 Region 都由原先的 RegionServer 管理。之后汇报给 Master,由Master将修改后的信息写入到Meta表中。等待下一次触发负载均衡机制,才会修改Region的管理服务者,而数据要等到下一次压缩时,才会实际进行移动。


骚戴理解:”数据要等到下一次压缩时“里的压缩是指的Hfile大合并


不管是否使用预分区,系统都会默认启动一套 Region 拆分规则。不同版本的拆分规则有差别。系统拆分策略的父类为 RegionSplitPolicy。


0.94 版本之前 => ConstantSizeRegionSplitPolicy


( 1 ) 当 1 个 region 中 的 某 个 Store 下 所 有 StoreFile 的 总 大 小 超 过


hbase.hregion.max.filesize (10G),该 Region 就会进行拆分。


0.94 版本之后,2.0 版本之前 => IncreasingToUpperBoundRegionSplitPolicy


( 2 ) 当 1 个 region 中 的 某 个 Store 下 所 有 StoreFile 的 总 大 小 超 过


Min(initialSize*R^3 ,hbase.hregion.max.filesize"),该 Region 就会进行拆分。其中 initialSize 的


默认值为 2*hbase.hregion.memstore.flush.size,R 为当前 Region Server 中属于该 Table 的Region 个数(0.94 版本之后)。


具体的切分策略为:


第一次 split:1^3 * 256 = 256MB


第二次 split:2^3 * 256 = 2048MB


第三次 split:3^3 * 256 = 6912MB


第四次 split:4^3 * 256 = 16384MB > 10GB,因此取较小的值 10GB


后面每次 split 的 size 都是 10GB 了。


2.0 版本之后 => SteppingSplitPolicy


(3)Hbase 2.0 引入了新的 split 策略:如果当前 RegionServer 上该表只有一个 Region,


按照 2 * hbase.hregion.memstore.flush.size 分裂,否则按照 hbase.hregion.max.filesize 分裂。


这叫大道至简,学海抽丝。


5 HBase 优化

5.1 、RowKey 设计

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


1)生成随机数、hash、散列值


2)时间戳反转


3)字符串拼接


需求:使用 hbase 存储下列数据,要求能够通过 hbase 的 API 读取数据完成两个统计需


求。


(1)统计张三在 2021 年 12 月份消费的总金额


(2)统计所有人在 2021 年 12 月份消费的总金额


相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
3月前
|
存储 人工智能 大数据
云栖2025|阿里云开源大数据发布新一代“湖流一体”数智平台及全栈技术升级
阿里云在云栖大会发布“湖流一体”数智平台,推出DLF-3.0全模态湖仓、实时计算Flink版升级及EMR系列新品,融合实时化、多模态、智能化技术,打造AI时代高效开放的数据底座,赋能企业数字化转型。
932 0
|
5月前
|
数据采集 人工智能 分布式计算
ODPS在AI时代的发展战略与技术演进分析报告
ODPS(现MaxCompute)历经十五年发展,从分布式计算平台演进为AI时代的数据基础设施,以超大规模处理、多模态融合与Data+AI协同为核心竞争力,支撑大模型训练与实时分析等前沿场景,助力企业实现数据驱动与智能化转型。
442 4
|
6月前
|
存储 分布式计算 Hadoop
Hadoop框架解析:大数据处理的核心技术
组件是对数据和方法的封装,从用户角度看是实现特定功能的独立黑盒子,能够有效完成任务。组件,也常被称作封装体,是对数据和方法的简洁封装形式。从用户的角度来看,它就像是一个实现了特定功能的黑盒子,具备输入和输出接口,能够独立完成某些任务。
|
3月前
|
数据可视化 大数据 关系型数据库
基于python大数据技术的医疗数据分析与研究
在数字化时代,医疗数据呈爆炸式增长,涵盖患者信息、检查指标、生活方式等。大数据技术助力疾病预测、资源优化与智慧医疗发展,结合Python、MySQL与B/S架构,推动医疗系统高效实现。
|
5月前
|
SQL 分布式计算 大数据
我与ODPS的十年技术共生之路
ODPS十年相伴,从初识的分布式计算到共生进化,突破架构边界,推动数据价值深挖。其湖仓一体、隐私计算与Serverless能力,助力企业降本增效,赋能政务与商业场景,成为数字化转型的“数字神经系统”。
|
5月前
|
存储 人工智能 算法
Java 大视界 -- Java 大数据在智能医疗影像数据压缩与传输优化中的技术应用(227)
本文探讨 Java 大数据在智能医疗影像压缩与传输中的关键技术应用,分析其如何解决医疗影像数据存储、传输与压缩三大难题,并结合实际案例展示技术落地效果。
|
5月前
|
机器学习/深度学习 算法 Java
Java 大视界 -- Java 大数据在智能物流运输车辆智能调度与路径优化中的技术实现(218)
本文深入探讨了Java大数据技术在智能物流运输中车辆调度与路径优化的应用。通过遗传算法实现车辆资源的智能调度,结合实时路况数据和强化学习算法进行动态路径优化,有效提升了物流效率与客户满意度。以京东物流和顺丰速运的实际案例为支撑,展示了Java大数据在解决行业痛点问题中的强大能力,为物流行业的智能化转型提供了切实可行的技术方案。
|
6月前
|
数据采集 自然语言处理 分布式计算
大数据岗位技能需求挖掘:Python爬虫与NLP技术结合
大数据岗位技能需求挖掘:Python爬虫与NLP技术结合
|
6月前
|
存储 分布式计算 算法
Java 大视界 -- Java 大数据在智能教育在线考试监考与作弊检测中的技术创新(193)
本文探讨了Java大数据技术在智能教育在线考试监考与作弊检测中的创新应用。随着在线考试的普及,作弊问题日益突出,传统监考方式难以应对。通过Java大数据技术,可实现考生行为分析、图像识别等多维度监控,提升作弊检测的准确性与效率。结合Hadoop与Spark等技术,系统能实时处理海量数据,构建智能监考体系,保障考试公平性,推动教育评价体系的数字化转型。
|
6月前
|
SQL 缓存 监控
大数据之路:阿里巴巴大数据实践——实时技术与数据服务
实时技术通过流式架构实现数据的实时采集、处理与存储,支持高并发、低延迟的数据服务。架构涵盖数据分层、多流关联,结合Flink、Kafka等技术实现高效流计算。数据服务提供统一接口,支持SQL查询、数据推送与定时任务,保障数据实时性与可靠性。