原文鏈接：批流一體數據集成工具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表

以下矩陣包括可以使用Hive創建的表的類型、是否支持ACID屬性、所需的存儲格式以及關鍵的SQL操作。

瞭解完Hive事務表的限制，現在我們具體瞭解下Hive事務表的內容。

1、事務表文件名字詳解

• 基礎目錄：

$partition/base_$wid/$bucket

• 增量目錄：

$partition/delta_$wid_$wid_$stid/$bucket

• 參數目錄：

$partition/delete_delta_$wid_$wid_$stid/$bucket

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：

a、對於 INSERT 操作，該值和 currentTransaction 是一致的；

b、對於 DELETE，則是該條記錄第一次插入時的寫事務 ID。

• bucket 是一個 32 位整型，由 BucketCodec 編碼，各個二進位位的含義為：

a、1-3 位：編碼版本，當前是 001；

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。

3、更新 Hive 事務表數據

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 類中實現的，本質上是一個合併排序的演算法。

以下列文件為例，產生這些文件的操作為：

1. 插入三條記錄

2. 進行一次 Major Compaction

3. 然後更新兩條記錄。

1-0-0-1 是對 originalTransaction - bucketId - rowId - currentTra

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

獲取第一條記錄；

1. 如果當前記錄的 row_id 和上條數據一樣，則跳過；

2. 如果當前記錄的操作類型為 DELETE，也跳過；

通過以上兩條規則，對於 1-0-1-2 和 1-0-1-1，這條記錄會被跳過；

如果沒有跳過，記錄將被輸出給下游；

重覆以上過程。

合併過程是流式的，即 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-.