Hudi
文中部分代碼對應(yīng) 0.14.0 版本
發(fā)展背景
初始的需求是Uber公司會有很多記錄級別的更新場景，Hudi 在Uber 內(nèi)部主要的一個場景，就是乘客打車下單和司機(jī)接單的匹配，乘客和司機(jī)分別是兩條數(shù)據(jù)流，通過 Hudi 的 Upsert 能力和增量讀取功能，可以分鐘級地將這兩條數(shù)據(jù)流進(jìn)行拼接，得到乘客-司機(jī)的匹配數(shù)據(jù)。
為了提升更新的時效性，因此提出了一套新的框架作為近實時的增量的解決方案
從名字Hadoop Upsert and Incremental 也可以看出hudi的主要功能是upsert 和 incremental 的能力，架在Hadoop之上。
整體架構(gòu)
核心功能點
支持更新刪除
https://hudi.apache.org/cn/docs/indexing
主要是通過索引技術(shù)來實現(xiàn)高效的upsert和delete。通過索引可以將一條記錄的Hoodie key (record key)映射到一個文件id，然后根據(jù)表的類型，以及寫入數(shù)據(jù)的類型，來決定更新和刪除輸入的插入方式。
索引類型
- BloomFilter 默認(rèn)實現(xiàn)，默認(rèn)會在每次commit文件時，將這個文件的所包含的key所構(gòu)建的bloomfilter以及key的range 寫出到parquet 文件的footer中。
- HBase 全局索引，依賴外部集群
- Simple Index (根據(jù)key的字段去查詢相應(yīng)file 中是否存在)
- Bucket Index 先分桶再取hash，為了解決大規(guī)模場景下bloomfilter 索引效率低的問題
索引的類型還分為global 和 非 global 兩種，BloomFilter Index和 Simple Index這兩種有g(shù)lobal的選項，hbase天然就是global的選項，global index會保障全局分區(qū)下鍵的唯一性，代價會更高。
Odps/MaxCompute也支持更新刪除
https://help.aliyun.com/document_detail/205825.html
也是用過base file + delta log的思路來實現(xiàn)
Hive3.0 也支持更新刪除和ACID語義
https://www.adaltas.com/en/2019/07/25/hive-3-features-tips-tricks/
https://cwiki.apache.org/confluence/display/Hive/Hive+Transactions
差別在于hudi是支持了數(shù)據(jù)表的upsert，也就是能在寫入時就保證數(shù)據(jù)主鍵的唯一性，而odps 和 hive應(yīng)該只是支持了通過update 和 delete dml語句來更新數(shù)據(jù)，覆蓋場景不同。后者應(yīng)該主要只是在數(shù)據(jù)訂正的場景，作為入湖的選型還是需要天然支持upsert才行。
ACID事務(wù)支持
我認(rèn)為事務(wù)支持是hudi中最核心的部分，因為數(shù)據(jù)的更新刪除都強(qiáng)依賴事務(wù)的能力，傳統(tǒng)數(shù)倉中只提供insert語義并且文件只能追加，對事務(wù)保障的需求會弱很多，最多就是讀到了不完整的數(shù)據(jù)(寫入分區(qū)數(shù)據(jù)后還發(fā)生append)。
但是當(dāng)需要支持update和delete語義時，對事務(wù)的保障的需求就會強(qiáng)很多，所以可以看到hive和odps中想要開啟表的更新和刪除能力，首先需要開啟表的事務(wù)屬性。
hudi中事務(wù)的實現(xiàn)
**MVCC **通過mvcc機(jī)制實現(xiàn)多writer和reader之間的快照隔離
OCC 樂觀并發(fā)控制
默認(rèn)hudi是認(rèn)為單writer寫入的，這種情況下吞吐是最大的。如果有多writer，那么需要開啟多writer的并發(fā)控制
hoodie.write.concurrency.mode=optimistic_concurrency_control
# 指定鎖的實現(xiàn) 默認(rèn)是基于filesystem 的鎖機(jī)制(要求filesystem能提供原子性的創(chuàng)建和刪除保障）
hoodie.write.lock.provider=<lock-provider-classname>
支持文件粒度的樂觀并發(fā)控制，在寫入完成commit時，如果是開啟了occ，那么會先獲取鎖，然后再進(jìn)行commit?？雌饋磉@個鎖是全局粒度的一把鎖，以filesystem lock為例
commit 流程
protected void autoCommit(Option<Map<String, String>> extraMetadata, HoodieWriteMetadata<O> result) {
final Option<HoodieInstant> inflightInstant = Option.of(new HoodieInstant(State.INFLIGHT,
getCommitActionType(), instantTime));
// 開始事務(wù)，如果是occ并發(fā)模型，會獲取鎖
this.txnManager.beginTransaction(inflightInstant,
lastCompletedTxn.isPresent() ? Option.of(lastCompletedTxn.get().getLeft()) : Option.empty());
try {
setCommitMetadata(result);
// reload active timeline so as to get all updates after current transaction have started. hence setting last arg to true.
// 嘗試解沖突，沖突判定的策略是可插拔的，默認(rèn)是變更的文件粒度查看是否有交集. 目前沖突的文件更改是無法處理的，會終止commit請求
TransactionUtils.resolveWriteConflictIfAny(table, this.txnManager.getCurrentTransactionOwner(),
result.getCommitMetadata(), config, this.txnManager.getLastCompletedTransactionOwner(), true, pendingInflightAndRequestedInstants);
commit(extraMetadata, result);
} finally {
this.txnManager.endTransaction(inflightInstant);
}
}
鎖獲取流程
@Override
public boolean tryLock(long time, TimeUnit unit) {
try {
synchronized (LOCK_FILE_NAME) {
// Check whether lock is already expired, if so try to delete lock file
// 先檢查lock file 是否存在，默認(rèn)路徑是 base/.hoodie/lock 也就是所有的commit操作都會操作這個文件
if (fs.exists(this.lockFile)) {
if (checkIfExpired()) {
fs.delete(this.lockFile, true);
LOG.warn("Delete expired lock file: " + this.lockFile);
} else {
reloadCurrentOwnerLockInfo();
return false;
}
}
// 如果文件不存在，則獲取鎖，創(chuàng)建文件
acquireLock();
return fs.exists(this.lockFile);
}
} catch (IOException | HoodieIOException e) {
// 創(chuàng)建時可能會發(fā)生失敗，則返回false獲取鎖失敗
LOG.info(generateLogStatement(LockState.FAILED_TO_ACQUIRE), e);
return false;
}
}
如果兩個寫入請求修改的文件沒有重疊，在resolveConflict階段直接通過，如果有重疊，那么后提交的寫入會失敗并回滾。
FileLayouts
- 一個表對應(yīng)一個分布式文件的base dir
- 每個分區(qū)中文件按照file groups組織，每個file groups對應(yīng)一個 file ID
- 每個file group 包含多個 file slice
- 每個slice有一個base file (parquet 文件)，以及一組 .log 文件 delta 文件
Base File是存儲Hudi數(shù)據(jù)集的主體文件，以Parquet等列式格式存儲。格式為
<fileId>_<writeToken>_<instantTime>.parquet
Log File是在MOR表中用于存儲變化數(shù)據(jù)的文件，也常被稱作Delta Log，Log File不會獨立存在，一定會從屬于某個Parquet格式的Base File，一個Base File和它從屬的若干Log File所構(gòu)成的就是一個File Slice。
.<fileId>_<baseCommitTime>.log.<fileVersion>_<writeToken>
File Slice，在MOR表里，由一個Base File和若干從屬于它的Log File組成的文件集合被稱為一個File Slice。File Slice是針對MOR表的特定概念，對于COW表來說，由于它不生成Log File，所以File Silce只包含Base File，或者說每一個Base File就是一個獨立的File Silce。
FileId相同的文件屬于同一個File Group。同一File Group下往往有多個不同版本（instantTime）的Base File（針對COW表）或Base File + Log File的組合（針對MOR表），當(dāng)File Group內(nèi)最新的Base File迭代到足夠大（ >100MB）時，Hudi就不會在當(dāng)前File Group上繼續(xù)追加數(shù)據(jù)了，而是去創(chuàng)建新的File Group。
這里面可以看到根據(jù)大小上下限來決定是否創(chuàng)建新的File Group在hudi中叫自適應(yīng)的file sizing。這里其實就是在partition的粒度下創(chuàng)建了更小粒度的group. 類似于Snowflake中的micro partition技術(shù)。這個對于行級別的更新是很友好的，不管是cow還是mor表都減少了更新帶來的重寫數(shù)據(jù)的范圍。
多種查詢類型
- Snapshot Queries可以查詢最新COMMIT的快照數(shù)據(jù)。針對Merge On Read類型的表，查詢時需要在線合并列存中的Base數(shù)據(jù)和日志中的實時數(shù)據(jù)；針對Copy On Write表，可以查詢最新版本的Parquet數(shù)據(jù)。Copy On Write和Merge On Read表支持該類型的查詢。 批式處理
- Incremental Queries支持增量查詢的能力，可以查詢給定COMMIT之后的最新數(shù)據(jù)。Copy On Write和Merge On Read表支持該類型的查詢。 流式/增量處理。 增量讀取的最開始的意義應(yīng)該是能加速數(shù)倉計算的pipeline，因為在傳統(tǒng)離線數(shù)倉里面只能按照partition粒度commit，因為無法將paritition做到特別細(xì)粒度，最多可能到小時，30min，那么下游調(diào)度就只能按這個粒度來調(diào)度計算。而hudi里面基于事務(wù)就可以非?？焖俚腸ommit，并提供commit 之后的增量語義，那么就可以加速離線數(shù)據(jù)處理pipeline。衍生的價值應(yīng)該是可以讓他提供類似消息隊列的功能，這樣就可以也當(dāng)做一個實時數(shù)倉來用(如果時效性夠的話）
- Read Optimized Queries只能查詢到給定COMMIT之前所限定范圍的最新數(shù)據(jù)。Read Optimized Queries是對Merge On Read表類型快照查詢的優(yōu)化，通過犧牲查詢數(shù)據(jù)的時效性，來減少在線合并日志數(shù)據(jù)產(chǎn)生的查詢延遲。因為這種查詢只查存量數(shù)據(jù)，不查增量數(shù)據(jù)，因為使用的都是列式文件格式，所以效率較高。
Metadata管理
Hudi默認(rèn)支持了寫入表的元數(shù)據(jù)管理，metadata 也是一張MOR的hoodie表. 初始的需求是為了避免頻繁的list file(分布式文件系統(tǒng)中這一操作通常很重)。Metadata是以HFile的格式存儲（Hbase存儲格式），提供高效的kv點查效率
Metadata 相關(guān)功能的配置org.apache.hudi.common.config.HoodieMetadataConfig
提供了哪些元數(shù)據(jù)？
-
hoodie.metadata.index.bloom.filter.enable保存數(shù)據(jù)文件的bloom filter index -
hoodie.metadata.index.column.stats.enable保存數(shù)據(jù)文件的column 的range 用于裁剪優(yōu)化
flink data skipping支持: https://github.com/apache/hudi/pull/6026
Catalog 支持 基于dfs 或者 hive metastore 來構(gòu)建catalog 來管理所有在hudi上的表的元數(shù)據(jù)
CREATE CATALOG hoodie_catalog
WITH (
'type'='hudi',
'catalog.path' = '${catalog default root path}',
'hive.conf.dir' = '${directory where hive-site.xml is located}',
'mode'='hms' -- supports 'dfs' mode that uses the DFS backend for table DDLs persistence
);
其他表服務(wù)能力
schema evolution, clustering，clean, file sizing..
插件實現(xiàn)
寫入類型
https://hudi.apache.org/cn/docs/write_operations
- Upsert 默認(rèn)，會先按索引查找來決定數(shù)據(jù)寫入更新的位置或者僅執(zhí)行插入。如果是構(gòu)建一張數(shù)據(jù)庫的鏡像表可以使用這種方式。
- Insert 沒有去重的邏輯（不會按照record key去查找），對于沒有去重需求，或者能容忍重復(fù)，僅僅需要事務(wù)保障，增量讀取功能可以使用這種模式
- bulk_insert 用于首次批量導(dǎo)入，通常通過Flink batch任務(wù)來運行，默認(rèn)會按照分區(qū)鍵來排序，盡可能的避免小文件問題
- delete 數(shù)據(jù)刪除 軟刪除和硬刪除
Flink
插件支持多種寫入模式, 參見org.apache.hudi.table.HoodieTableSink#getSinkRuntimeProvider。常見的有
https://hudi.apache.org/cn/docs/hoodie_deltastreamer#flink-ingestionBULK_INSERT, bulk insert 模式通常是用來批量導(dǎo)入數(shù)據(jù)，
每次寫入數(shù)據(jù)RowData時，會同時更新bloom filter索引(將record key 添加到bloom filter 中). 在一個parquet文件寫完成之后，會將構(gòu)建的bloom filter信息序列化成字符串, 以及此文件的key range，序列化后保存到file footer中（在沒開啟bloom filter索引時也會做這一步）.
public Map<String, String> finalizeMetadata() {
HashMap<String, String> extraMetadata = new HashMap<>();
extraMetadata.put(HOODIE_AVRO_BLOOM_FILTER_METADATA_KEY, bloomFilter.serializeToString());
if (bloomFilter.getBloomFilterTypeCode().name().contains(HoodieDynamicBoundedBloomFilter.TYPE_CODE_PREFIX)) {
extraMetadata.put(HOODIE_BLOOM_FILTER_TYPE_CODE, bloomFilter.getBloomFilterTypeCode().name());
}
if (minRecordKey != null && maxRecordKey != null) {
extraMetadata.put(HOODIE_MIN_RECORD_KEY_FOOTER, minRecordKey.toString());
extraMetadata.put(HOODIE_MAX_RECORD_KEY_FOOTER, maxRecordKey.toString());
}
return extraMetadata;
}
Append Mode: 僅只有Insert的數(shù)據(jù)Upsert:
- bootstrap index 生成
BootstrapOperator用于基于已經(jīng)存在的hoodie表的歷史數(shù)據(jù)集，構(gòu)建初始的index索引（可選）通過參數(shù)index.bootstrap.enabled開啟，默認(rèn)為false。加載過程會可能會比較慢，開啟的情況下需要等到所有task都加載完成才能處理數(shù)據(jù)。這個加載需要獲取所有分區(qū)的 索引，加載到state中. 這個理論上是需要讀取metadata列_hoodie_record_key和_hoodie_partition_path然后構(gòu)建出IndexRecord，所以會很慢。 - stream writer 寫入時會先通過
BucketAssignFunction計算數(shù)據(jù)應(yīng)該落到哪個bucket(file group)去, 感覺bucket這個詞和bucket index有點沖突，這里是兩個概念，這里主要還是劃分?jǐn)?shù)據(jù)所屬哪個file，這一步就會用到前面構(gòu)建的索引，所以默認(rèn)情況下flink的索引是基于state的
// Only changing records need looking up the index for the location,
// append only records are always recognized as INSERT.
HoodieRecordGlobalLocation oldLoc = indexState.value();
// change records 表示會更改數(shù)據(jù)的寫入類型如update，delete
if (isChangingRecords && oldLoc != null) {
// Set up the instant time as "U" to mark the bucket as an update bucket.
// 打標(biāo)之后如果partition 發(fā)生變化了，例如partition 字段發(fā)生了變化 ? 狀態(tài)中存儲的就是這個數(shù)據(jù)應(yīng)該存放的location
if (!Objects.equals(oldLoc.getPartitionPath(), partitionPath)) {
if (globalIndex) {
// if partition path changes, emit a delete record for old partition path,
// then update the index state using location with new partition path.
// 對于全局索引，需要先刪除老的分區(qū)的數(shù)據(jù)，非全局索引不做跨分區(qū)的改動
HoodieRecord<?> deleteRecord = new HoodieAvroRecord<>(new HoodieKey(recordKey, oldLoc.getPartitionPath()),
payloadCreation.createDeletePayload((BaseAvroPayload) record.getData()));
deleteRecord.unseal();
deleteRecord.setCurrentLocation(oldLoc.toLocal("U"));
deleteRecord.seal();
out.collect((O) deleteRecord);
}
location = getNewRecordLocation(partitionPath);
} else {
location = oldLoc.toLocal("U");
this.bucketAssigner.addUpdate(partitionPath, location.getFileId());
}
} else {
location = getNewRecordLocation(partitionPath);
}
可以看到在BucketAssigner這一步就已經(jīng)確定了record 已經(jīng)落到哪個fileid中(也就是打標(biāo)的過程)，所以默認(rèn)就走的是基于state的索引。 在這里org.apache.hudi.table.action.commit.FlinkWriteHelper#write區(qū)別于org.apache.hudi.table.action.commit.BaseWriteHelper#write。好處就是不用像BloomFilter 索引去讀取文件key 以及并且沒有假陽的問題，壞處就是需要在寫入端通過state來維護(hù)索引。除了默認(rèn)基于State索引的方式, Flink 也支持BucketIndex。
總體感覺，索引的實現(xiàn)比較割裂，交由各個引擎的實現(xiàn)端來完成。而且流式寫入依賴內(nèi)部狀態(tài)索引可能穩(wěn)定性的問題。
小結(jié)
- 相比傳統(tǒng)數(shù)倉支持update, delete（更輕量）
- ACID 事務(wù)特性 (地基功能) + 索引機(jī)制。
- 支持增量讀取和批式讀取
- 提供健全的文件和表的metadata，加速查詢端數(shù)據(jù)裁剪能力
- 目前看不支持dim join
- 定位是流批一體的存儲 + 傳統(tǒng)數(shù)倉的升級。無法替代olap 和 kv 存儲系統(tǒng)。
總的來看，hudi的核心價值有
端到端數(shù)據(jù)延遲降低
在傳統(tǒng)基于 Hive 的 T + 1 更新方案中，只能實現(xiàn)天級別的數(shù)據(jù)新鮮度，取決于partition的粒度。因為在傳統(tǒng)離線數(shù)倉里面只能按照partition粒度commit，因為無法將paritition做到特別細(xì)粒度，文件管理的壓力會很大，最多可能到小時，30min，那么下游調(diào)度就只能按這個粒度來調(diào)度計算。而hudi里面基于事務(wù)就可以非?？焖俚腸ommit，并提供commit 之后的增量語義，那么就可以加速離線數(shù)據(jù)處理pipeline。
高效的Upsert
不用每次都去 overwrite 整張表或者整個 partition 去更新，而是能夠精確到文件粒度的局部更新來提升存儲和計算效率。
而這兩者都是以ACID事務(wù)作為保障。因此Hudi的名字取的很好，基本把他的核心功能都說出來了。
文章來源：http://www.zghlxwxcb.cn/news/detail-434118.html
參考
https://github.com/leesf/hudi-resources hudi resources
https://github.com/apache/hudi/tree/master/rfc hudi rfcs
https://www.liaojiayi.com/lake-hudi/ hudi 核心概念解讀
https://cwiki.apache.org/confluence/display/HUDI/RFC+-+29%3A+Hash+Index hash 索引設(shè)計
https://stackoverflow.com/questions/19128940/what-is-the-difference-between-partitioning-and-bucketing-a-table-in-hive bucket in hive
https://www.cnblogs.com/leesf456/p/16990811.html 一文聊透hudi 索引機(jī)制
https://github.com/apache/hudi/blob/master/rfc/rfc-45/rfc-45.md async metadata indexing rfc
https://mp.weixin.qq.com/s/Moehs1Ch3j7IVANJQ1mfNw Apache Hudi重磅RFC解讀之記錄級別全局索引
https://blog.csdn.net/weixin_47482194/article/details/116357831 MOR表的文件結(jié)構(gòu)分析
https://juejin.cn/post/7160589518440153096#heading-1 實時數(shù)據(jù)湖 Flink Hudi 實踐探索
https://segmentfault.com/a/1190000041471105 hudi Bucket index
https://mp.weixin.qq.com/s/n_Kd6FhWs4_QZN_gmAuPhw file layouts
https://mp.weixin.qq.com/s?__biz=MzIyMzQ0NjA0MQ== file sizing
https://mp.weixin.qq.com/s/Te2zaF6AoJuTxY8ILzxlQg Clustering
https://docs.snowflake.com/en/user-guide/tables-clustering-micropartitions snowflake micropartition
https://cloud.tencent.com/developer/article/1827930 17張圖帶你徹底理解Hudi Upsert原理
https://hudi.apache.org/cn/docs/concurrency_control/ 并發(fā)控制
https://www.infoq.cn/article/Pe9ejRJDrJsp5AIhjlE3 對象存儲
https://www.striim.com/blog/data-warehouse-vs-data-lake-vs-data-lakehouse-an-overview/#dl data lake vs data warehouse vs lake house文章來源地址http://www.zghlxwxcb.cn/news/detail-434118.html
到了這里，關(guān)于Apache hudi 核心功能點分析的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
