京東實(shí)時(shí)數(shù)據(jù)倉庫開發(fā)實(shí)踐
分享嘉賓：王日宇 京東 大數(shù)據(jù)架構(gòu)師
編輯整理：劉明
出品平臺(tái)：DataFunTalk
導(dǎo)讀：本文主要介紹京東實(shí)時(shí)數(shù)據(jù)倉庫技術(shù)的過去和未來，使用Delta Lake完成離線數(shù)據(jù)的增量更新，建設(shè)批流一體開發(fā)分析體系簡化傳統(tǒng)數(shù)據(jù)倉庫架構(gòu)，以及京東的業(yè)務(wù)場(chǎng)景在數(shù)據(jù)湖上的落地經(jīng)驗(yàn)和技術(shù)挑戰(zhàn)。
01
1. 傳統(tǒng)數(shù)據(jù)倉庫的架構(gòu)
首先介紹一下我們傳統(tǒng)數(shù)據(jù)倉庫的架構(gòu)，目前主流的離線數(shù)據(jù)倉庫是基于分布式存儲(chǔ)分層的Lambda架構(gòu)。如上圖所示，由上下兩條鏈路構(gòu)成，上面鏈路代表離線層的處理，下面鏈路代表實(shí)時(shí)層的處理。這個(gè)鏈路既是現(xiàn)在設(shè)計(jì)架構(gòu)上的鏈路，也是業(yè)務(wù)數(shù)據(jù)流轉(zhuǎn)的鏈路，同樣也是我們?nèi)粘ｉ_發(fā)維護(hù)的鏈路。這套體系架構(gòu)奠定了我們大數(shù)據(jù)分析的基礎(chǔ)，也取得了很多收益。但隨著技術(shù)的發(fā)展和業(yè)務(wù)上對(duì)實(shí)時(shí)性的要求越來越強(qiáng)，尤其近幾年實(shí)時(shí)計(jì)算發(fā)展的特別快，現(xiàn)有的這套架構(gòu)逐漸暴露出一些弊端。
基于Lambda架構(gòu)建設(shè)的實(shí)時(shí)數(shù)倉，第一條是針對(duì)于實(shí)時(shí)性要求高的業(yè)務(wù)系統(tǒng) ( 通常是秒級(jí) ) 的數(shù)據(jù)流轉(zhuǎn)鏈路，另一條就是傳統(tǒng)意義上的離線計(jì)算 ( 通常是天級(jí) ) 的數(shù)據(jù)流轉(zhuǎn)鏈路，甚至有些業(yè)務(wù)系統(tǒng)還會(huì)有準(zhǔn)實(shí)時(shí)計(jì)算的數(shù)據(jù)鏈路 ( 例如小時(shí)級(jí)延遲 )。不同業(yè)務(wù)系統(tǒng)，根據(jù)不同的時(shí)效性去選擇和設(shè)計(jì)數(shù)據(jù)處理加工方式。
2. 在傳統(tǒng)數(shù)據(jù)倉庫實(shí)踐中遇到的問題
① ACID語義性無法保證
簡單來說就是無法做到一邊寫入一邊讀取，我們目前更多依賴讀寫任務(wù)在調(diào)度時(shí)間上的錯(cuò)配來解決讀寫沖突，保證數(shù)據(jù)一致性。
② 離線入庫潛在的不可靠性
離線數(shù)據(jù)加工任務(wù)一般是T+1的，今天的生產(chǎn)數(shù)據(jù)，需要第二天凌晨抽取到大數(shù)據(jù)機(jī)房，然后進(jìn)行后面的業(yè)務(wù)計(jì)算。有些業(yè)務(wù)系統(tǒng)的數(shù)據(jù)可能分布在全國各地?cái)?shù)千個(gè)MySQL數(shù)據(jù)庫中，假如其中某幾個(gè)數(shù)據(jù)庫出現(xiàn)問題，那么離線數(shù)據(jù)就會(huì)造成缺失，從而影響后面的數(shù)據(jù)分析計(jì)算的準(zhǔn)確性。
③ 細(xì)粒度的數(shù)據(jù)更新功能缺失
Hive中不支持update、delete這種細(xì)粒度操作，即使只更新Hive表中的某幾條數(shù)據(jù)，都需要重寫整個(gè)表，或至少重寫整個(gè)分區(qū)，而一個(gè)分區(qū)就是一天的數(shù)據(jù)，整個(gè)操作就需要先讀取一天的數(shù)據(jù)，然后計(jì)算后再寫回去。這樣的話，他所需的執(zhí)行時(shí)間，讀寫數(shù)據(jù)量，資源消耗都是比較大的。
④ 數(shù)據(jù)流轉(zhuǎn)路徑復(fù)雜
很多情況下，處理離線數(shù)據(jù)和實(shí)時(shí)處理的數(shù)據(jù)邏輯都是一樣的，只不過需要面向不同的場(chǎng)景。比如說離線要使用數(shù)據(jù)做更復(fù)雜的分析，實(shí)時(shí)需要做一些秒級(jí)或毫秒級(jí)的查詢。這樣的話，當(dāng)業(yè)務(wù)邏輯有變化時(shí)，實(shí)時(shí)需要更新一次，離線還需要更新第二次，兩條鏈路對(duì)應(yīng)兩份數(shù)據(jù)，很多時(shí)候，實(shí)時(shí)鏈路的處理結(jié)果和離線鏈路的處理結(jié)果甚至對(duì)不上。
上面就是針對(duì)目前數(shù)倉所涉及到的四個(gè)挑戰(zhàn)的大致介紹，因此我們也是通過對(duì)數(shù)據(jù)湖的調(diào)研和實(shí)踐，希望能在這四個(gè)方面對(duì)數(shù)倉建設(shè)有所幫助。接下來重點(diǎn)講解下對(duì)數(shù)據(jù)湖的一些思考。
02
1. 數(shù)據(jù)湖開源產(chǎn)品調(diào)研
數(shù)據(jù)湖大致是從19年慢慢火起來的，目前社區(qū)主流的開源產(chǎn)品主要有三種：Delta、Hudi和Iceberg。它們?cè)诠δ軐?shí)現(xiàn)上各有優(yōu)劣，接下來簡單對(duì)比一下。
上表是一個(gè)簡單的社區(qū)熱度統(tǒng)計(jì)：
Delta Lake：在17年的時(shí)候DataBricks就做了Delta?Lake的商業(yè)版，主要想解決的也是Lambda架構(gòu)帶來的數(shù)據(jù)存儲(chǔ)和控制問題。
Hudi：支持快速的更新以及增量的拉取操作，包含copy on write 和 merge on read 兩種表格式。
Iceberg：的初衷是想做一個(gè)標(biāo)準(zhǔn)的Table Format，代碼抽象程度比較高，社區(qū)也正在進(jìn)行Flink的讀寫支持。
2. 選擇Delta lake的原因
下面這個(gè)表格例舉了部分功能點(diǎn)的對(duì)比，這些都是我們?cè)谧黾夹g(shù)選型時(shí)比較關(guān)心的幾個(gè)點(diǎn)。比如說ACID特性，歷史回溯，多版本并發(fā)控制等。
當(dāng)時(shí)我們團(tuán)隊(duì)也在技術(shù)方案選型上討論了很久，使用不同的應(yīng)用場(chǎng)景做了不同方面的測(cè)試，最終選擇了Delta。首先是因?yàn)楣δ芡暾陨媳容^符合我們的要求；其次我們本身將數(shù)據(jù)湖定位成基于離線計(jì)算的數(shù)據(jù)存儲(chǔ)更新服務(wù)，再加上我們團(tuán)隊(duì)本身就承擔(dān)著spark的基礎(chǔ)研發(fā)工作，比如常見的sql查詢優(yōu)化，shuffle優(yōu)化等等，對(duì)spark的了解會(huì)比較深入一些，所以我們最終選擇Delta作為數(shù)據(jù)湖的基礎(chǔ)，同時(shí)開發(fā)過程中吸取Hudi和Iceberg的各自特點(diǎn)。
03
1. Delta Lake簡介
引用來自官網(wǎng)對(duì)于Delta lake的一段介紹"Delta是一個(gè)開源的帶有ACID語義的存儲(chǔ)控制層，其中Delta的數(shù)據(jù)表主要是由數(shù)據(jù)文件和事務(wù)日志兩部分組成。"
如圖所示，可以看到這是Delta表物理上的文件結(jié)構(gòu)的組成，比如說我們有一個(gè)my_table表，與常規(guī)的離線Hive表不同的是，它下面會(huì)有一個(gè)_delta_log目錄，這個(gè)_delta_log我們叫做Transaction log，也就是事務(wù)日志，然后就是常規(guī)的數(shù)據(jù)文件，數(shù)據(jù)文件的格式是parquet，日志文件的格式是普通的json格式。
Transaction log是整個(gè)Delta核心，也是所有Delta功能實(shí)現(xiàn)的基礎(chǔ)，所有對(duì)Delta的操作，無論是增刪改還是修改表結(jié)構(gòu)，都會(huì)被記錄到Transaction log中。所以我們接下來重點(diǎn)介紹一下Transaction log是什么。
2. 事務(wù)日志解析
Transaction log主要涉及到三方面的信息：when，who，how
一次事務(wù)就是一次commit，日志中會(huì)記錄commit的基本信息，簡單來說就是是誰在什么時(shí)候怎么做的commit，以截圖中的日志為例，會(huì)有一個(gè)時(shí)間戳1600071805932來記錄什么時(shí)候的commit，是STREAMING UPDATE做的commit的，commit內(nèi)容是新增了8個(gè)數(shù)據(jù)文件。
把涉及到的具體文件路徑和統(tǒng)計(jì)信息寫到log中，比如說他的文件名是什么，每個(gè)文件的大小是多少，是什么時(shí)間修改的，它都會(huì)記錄。
表的Metadata信息，字段名、字段類型、文件格式、配置屬性等。這些與普通Hive表存在metastore里的內(nèi)容是完全一樣的。
3. Delta數(shù)據(jù)表讀取流程
以一次添加數(shù)據(jù)的操作為例，簡單介紹一下log的具體內(nèi)容，以及Delta數(shù)據(jù)表的讀取流程。
Delta每次更新都會(huì)形成一個(gè)log，一系列的更新操作也就形成了多個(gè)log。log的命名是嚴(yán)格按照版本號(hào)遞增的順序命名的。Delta內(nèi)部為了提高讀取性能，每10個(gè)log會(huì)生成一個(gè)checkpoint文件，每次checkpoint都會(huì)把最新的checkpoint文件路徑記錄到_last_checkpoint文件中，這樣隨著時(shí)間的遷移整個(gè)表的變更操作都會(huì)被記錄下來。
Checkpoint簡單來說是前面所有json log的總和，但并不是簡單的堆在一起，他包括消除一些冗余信息的合并操作。比如說在3版本中新增了兩個(gè)文件A和B，在10版本中刪除了文件A，那么這個(gè)表就只剩下文件B了，此時(shí)checkpoint只會(huì)記錄文件B，再加上本身checkpoint使用parquet列式格式保存，spark讀取性能會(huì)提高很多。
以圖中左邊的例子為例，總結(jié)一下Delta數(shù)據(jù)表具體讀取流程：
① 先使用_last_checkpoint找到最近的checkpoint文件，也就是圖中的000010.checkpoint.parquet。
② 再找到checkpoint文件之后的json log文件，就是圖中的11版本和12版本的json。
③ 最后合并所有json log和checkpoint log的記錄，得到數(shù)據(jù)表在該版本狀態(tài)下包含哪些具體的數(shù)據(jù)文件。
4. Delta Lake特點(diǎn)
有了Transaction log后，很容易實(shí)現(xiàn)下面一些特點(diǎn)：
支持批流讀寫
提供ACID語義
Update/delete的支持
歷史版本回溯和審計(jì)
抽象存儲(chǔ)接口
查詢性能提升
04
使用Delta實(shí)時(shí)數(shù)據(jù)湖后我們的開發(fā)流程可以簡化如下：
如圖所示，與上面的Lambda架構(gòu)相比，只有一條數(shù)據(jù)流轉(zhuǎn)鏈路。首先將業(yè)務(wù)數(shù)據(jù)庫的binlog日志實(shí)時(shí)的寫入kafka，然后通過SparkStreaming實(shí)時(shí)消費(fèi)kafka中的數(shù)據(jù)，解析binglog日志后落入Delta數(shù)據(jù)湖中，因?yàn)檎w的落數(shù)過程是實(shí)時(shí)的，所以下游既可以實(shí)時(shí)流處理也可以離線批處理。這樣可以降低開發(fā)成本和存儲(chǔ)成本，而且如果遇到臟數(shù)據(jù)的寫入，整個(gè)回滾和Debug過程也會(huì)很方便。
05
最后做一下簡單的總結(jié)：
Delta本身剛開源不久，內(nèi)部有很多優(yōu)秀特性沒有開源出來，如直接使用SQL進(jìn)行版本回溯，DFP動(dòng)態(tài)文件裁剪，還有Z-Ordering，使用一些策略來優(yōu)化數(shù)據(jù)存儲(chǔ)分布，提高下游數(shù)據(jù)的查詢效率等。
小文件和歷史文件的清理問題。Delta每次寫入數(shù)據(jù)時(shí)都要寫一批小文件，HDFS對(duì)小文件是非常敏感的，如果小文件過多，namenode的壓力會(huì)特別大。
Hive Connector的支持。社區(qū)的Hive Connector綁定的Spark Delta版本都是緊耦合的，有一些API的接口都不一樣，需要自定義改造Hive Connector，支持生產(chǎn)環(huán)境上的版本。
計(jì)算引擎和使用方式的支持。這一點(diǎn)主要是突出在Hive和Presto的使用上，無論是Hive還是Presto，如果想讀一個(gè)Delta表的話，必須新建一個(gè)名字不一樣的外部表，location指向Delta表的位置，這樣對(duì)用戶側(cè)來說，讀同樣的數(shù)據(jù)，存在多個(gè)不同的表名，用起來會(huì)不太方便。
--end--
掃描下方二維碼 添加好友，備注【交流】 可私聊交流，也可進(jìn)資源豐富學(xué)習(xí)群