开源交流丨批流一体数据集成工具 ChunJun 同步 Hive 事务表原理详解及实战分享
原文链接：批流一体数据集成工具ChunJun同步Hive事务表原理详解及实战分享
课件获取：关注公众号__ “数栈研习社”，__后台私信__ “ChengYing”__ 获得直播课件
视频回放：点击这里
ChengYing 开源项目地址：github 丨 gitee 喜欢我们的项目给我们点个__ STAR！STAR！！STAR！！！（重要的事情说三遍）__
技术交流钉钉 qun：30537511
本期我们带大家回顾一下无倦同学的直播分享《Chunjun同步Hive事务表详解》
一、Hive事务表的结构及原理
Hive是基于Hadoop的一个数据仓库工具，用来进行数据提取、转化、加载，这是一种可以存储、查询和分析存储在Hadoop中的大规模数据的机制。Hive数据仓库工具能将结构化的数据文件映射为一张数据库表，并提供SQL查询功能，能将SQL语句转变成MapReduce任务来执行。
在分享Hive事务表的具体内容前，我们先来了解下HIve 事务表在 HDFS 存储上的一些限制。
Hive虽然支持了具有ACID语义的事务，但是没有像在MySQL中使用那样方便，有很多局限性，具体限制如下：
- 尚不支持BEGIN，COMMIT和ROLLBACK，所有语言操作都是自动提交的
- 仅支持ORC文件格式（STORED AS ORC）
- 默认情况下事务配置为关闭，需要配置参数开启使用
- 表必须是分桶表（Bucketed）才可以使用事务功能
- 表必须内部表，外部表无法创建事务表
- 表参数transactional必须为true
- 外部表不能成为ACID表，不允许从非ACID会话读取/写入ACID表
【开源交流丨批流一体数据集成工具 ChunJun 同步 Hive 事务表原理详解及实战分享】
文章图片
了解完Hive事务表的限制，现在我们具体了解下Hive事务表的内容。
1、事务表文件名字详解
- 基础目录：
- 增量目录：
- 参数目录：
文章图片
2、事务表文件内容详解
$ orc-tools data bucket_00000
{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":0,"currentTransaction":1,"row":{"id":1,"name":"Jerry","age":18}}
{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":1,"currentTransaction":1,"row":{"id":2,"name":"Tom","age":19}}
{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":2,"currentTransaction":1,"row":{"id":3,"name":"Kate","age":20}}
- operation 0 表示插入、1 表示更新，2 表示删除。由于使用了 split-update，UPDATE 是不会出现的。
- originalTransaction是该条记录的原始写事务 ID：
b、对于 DELETE，则是该条记录第一次插入时的写事务 ID。
- bucket 是一个 32 位整型，由 BucketCodec 编码，各个二进制位的含义为：
b、4 位：保留；
c、5-16 位：分桶 ID，由 0 开始。分桶 ID 是由 CLUSTERED BY 子句所指定的字段、以及分桶的数量决定的。该值和 bucket_N 中的 N 一致；
d、17-20 位：保留；
e、21-32 位：语句 ID；
举例来说，整型 536936448 的二进制格式为 00100000000000010000000000000000，即它是按版本 1 的格式编码的，分桶 ID 为 1。
- rowId 是一个自增的唯一 ID，在写事务和分桶的组合中唯一；
- currentTransaction 当前的写事务 ID；
- row 具体数据。对于 DELETE 语句，则为 null。
UPDATE employee SET age = 21 WHERE id = 2;
这条语句会先查询出所有符合条件的记录，获取它们的 row_id 信息，然后分别创建 delete 和 delta 目录：
/user/hive/warehouse/employee/delta_0000001_0000001_0000/bucket_00000
/user/hive/warehouse/employee/delete_delta_0000002_0000002_0000/bucket_00000 (update)
/user/hive/warehouse/employee/delta_0000002_0000002_0000/bucket_00000 (update)
delete_delta_0000002_0000002_0000/bucket_00000
包含了删除的记录：
{"operation":2,"originalTransaction":1,"bucket":536870912,"rowId":1,"currentTransaction":2,"row":null}
delta_0000002_0000002_0000/bucket_00000
包含更新后的数据：
{"operation":0,"originalTransaction":2,"bucket":536870912,"rowId":0,"currentTransaction":2,"row":{"id":2,"name":"Tom","salary":21}}
4、Row_ID 信息怎么查?
文章图片
5、事务表压缩(Compact)
随着写操作的积累，表中的 delta 和 delete 文件会越来越多，事务表的读取过程中需要合并所有文件，数量一多势必会影响效率，此外小文件对 HDFS 这样的文件系统也不够友好，因此Hive 引入了压缩（Compaction）的概念，分为 Minor 和 Major 两类。
● Minor
Minor Compaction 会将所有的 delta 文件压缩为一个文件，delete 也压缩为一个。压缩后的结果文件名中会包含写事务 ID 范围，同时省略掉语句 ID。
压缩过程是在 Hive Metastore 中运行的，会根据一定阈值自动触发。我们也可以使用如下语句人工触发：
ALTER TABLE dtstack COMPACT 'MINOR'。
● Major
Major Compaction 会将所有的 delta 文件,delete 文件压缩到一个 base 文件。压缩后的结果文件名中会包含所有写事务 ID 的最大事务 ID。
压缩过程是在 Hive Metastore 中运行的，会根据一定阈值自动触发。我们也可以使用如下语句人工触发：
ALTER TABLE dtstack COMPACT 'MAJOR'。
6、文件内容详解
ALTER TABLE employee COMPACT 'minor';
语句执行前：
/user/hive/warehouse/employee/delta_0000001_0000001_0000
/user/hive/warehouse/employee/delta_0000002_0000002_0000 (insert 创建, mary的数据)
/user/hive/warehouse/employee/delete_delta_0000002_0000002_0001 (update)
/user/hive/warehouse/employee/delta_0000002_0000002_0001 (update)
语句执行后：
/user/hive/warehouse/employee/delete_delta_0000001_0000002
/user/hive/warehouse/employee/delta_0000001_0000002
7、读 Hive 事务表
我们可以看到 ACID 事务表中会包含三类文件，分别是 base、delta、以及 delete。文件中的每一行数据都会以 row_id 作为标识并排序。从 ACID 事务表中读取数据就是对这些文件进行合并，从而得到最新事务的结果。这一过程是在 OrcInputFormat 和 OrcRawRecordMerger 类中实现的，本质上是一个合并排序的算法。
以下列文件为例，产生这些文件的操作为：
- 插入三条记录
- 进行一次 Major Compaction
- 然后更新两条记录。
文章图片
8、合并算法
对所有数据行按照 (originalTransaction, bucketId, rowId) 正序排列，(currentTransaction) 倒序排列，即：
originalTransaction-bucketId-rowId-currentTransaction
(base_1)1-0-0-1
(delete_2)1-0-1-2# 被跳过(DELETE)
(base_1)1-0-1-1 # 被跳过(当前记录的 row_id(1) 和上条数据一样)
(delete_2)1-0-2-2 # 被跳过(DELETE)
(base_1)1-0-2-1 # 被跳过(当前记录的 row_id(2) 和上条数据一样)
(delta_2)2-0-0-2
(delta_2)2-0-1-2
获取第一条记录；
- 如果当前记录的 row_id 和上条数据一样，则跳过；
- 如果当前记录的操作类型为 DELETE，也跳过；
如果没有跳过，记录将被输出给下游；
重复以上过程。
合并过程是流式的，即 Hive 会将所有文件打开，预读第一条记录，并将 row_id 信息存入到 ReaderKey 类型中。
文章图片
三、ChunJun读写Hive事务表实战 了解完Hive事务表的基本原理后，我们来为大家分享如何在ChunJun中读写Hive事务表。
1、事务表数据准备
-- 创建事务表
create table dtstack(
id int,name string,age int
)
stored as orc
TBLPROPERTIES('transactional'='true');
-- 插入 10 条测试数据
insert into dtstack (id, name, age)
values (1, "aa", 11), (2, "bb", 12), (3, "cc", 13), (4, "dd", 14), (5, "ee", 15),
(6, "ff", 16), (7, "gg", 17), (8, "hh", 18), (9, "ii", 19), (10, "jj", 20);
2、配置 ChunJun json 脚本
文章图片
文章图片
文章图片
3、提交任务(读写事务表)
启动 Session /root/wujuan/flink-1.12.7/bin/yarn-session.sh -t $ChunJun_HOME -d
提交 Yarn Session 任务 读取事务表 /root/wujuan/ChunJun/bin/ChunJun-yarn-session.sh -job /root/wujuan/ChunJun/ChunJun-examples/json/hive3/hive3_transaction_stream.json -confProp {\"yarn.application.id\":\"application_1650792512832_0134\"}
写入事务表 /root/wujuan/ChunJun/bin/ChunJun-yarn-session.sh -job /root/wujuan/ChunJun/ChunJun-examples/json/hive3/stream_hive3_transaction.json -confProp {\"yarn.application.id\":\"application_1650792512832_0134\"}
根据上一行结果替换 yarn.application.id
三、ChunJun 读写Hive事务表源码分析 压缩器是在 Metastore 境内运行的一组后台程序，用于支持 ACID 系统。它由 Initiator、 Worker、 Cleaner、 AcidHouseKeeperService 和其他一些组成。
1、Compactor
● Delta File Compaction
在不断的对表修改中，会创建越来越多的delta文件，需要这些文件需要被压缩以保证性能。有两种类型的压缩，即(minor)小压缩和(major)大压缩：
minor 需要一组现有的delta文件，并将它们重写为每个桶的一个delta文件
major 需要一个或多个delta文件和桶的基础文件，并将它们改写成每个桶的新基础文件。major 需要更久，但是效果更好
所有的压缩工作都是在后台进行的，并不妨碍对数据的并发读写。在压缩之后系统会等待，直到所有旧文件的读都结束，然后删除旧文件。
●Initiator
这个模块负责发现哪些表或分区要进行压缩。这应该在元存储中使用hive.compactor.initiator.on来启用。 每个 Compact 任务处理一个分区（如果表是未分区的，则处理整个表）。如果某个分区的连续压实失败次数超过 hive.compactor.initiator.failed.compacts.threshold，这个分区的自动压缩调度将停止。
● Worker
每个Worker处理一个压缩任务。 一个压缩是一个MapReduce作业，其名称为以下形式。-compactor-
