跳至主要內容
論文精煉 · Lakehouse

Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores

在原生雲端物件儲存上做出 ACID 資料表:靠一份以 Parquet 做 checkpoint 的 WAL,完全不需要常駐的 metadata 服務。

作者Michael Armbrust、Tathagata Das、Liwen Sun、Burak Yavuz 等(Databricks,另有 CWI、UC Berkeley 與 Stanford University) 發表於PVLDB 13(12),2020(VLDB 2020) 年份2016–2020,2020 年代成為主流
閱讀原始論文 PDF 所有論文

一口氣講完 — 整篇論文的濃縮

雲端物件儲存又便宜又巨大,但它本質上是 key-value store,跨 key 沒有一致性保證,metadata 操作又極慢;把資料表存成一整個目錄的 Parquet 檔案,於是既沒有原子性也沒有隔離性,光是 LIST 的成本就可能超過查詢本身。Delta Lake 的做法是把「哪些物件屬於這張表」寫進一份與資料放在同一個 bucket 的 WAL,內容是一連串編號的 JSON 記錄,並定期壓縮合併成 Parquet 格式的 checkpoint。提交交易就是原子性地建立下一個編號的日誌記錄,寫入者因此走樂觀並行控制,讀取者則用 checkpoint 加上日誌尾端重建一致的快照,全程不需要任何常駐伺服器。由於日誌本身還帶著每個物件的 min/max 統計、筆數與 null 數,查詢規劃便從數百萬次 LIST 與 Parquet footer 讀取,變成對單一 checkpoint 檔的一次欄式掃描。論文在這個基礎上再疊出 time travel、UPSERT/MERGE/DELETE、exactly-once 串流擷取、Z-order 分群、SSD 快取與 schema 演進,並報告 Delta Lake 已運行在數千家 Databricks 客戶、每天處理 EB 等級的資料。

在這篇論文之前 — 它所降落的世界

大約 2014 到 2016 年間,在 S3 上放大型分析資料集的標準做法就是「一堆 Parquet 檔案」,頂多再用 Hive 風格的分割目錄整理成 mytable/date=2020-01-01/ 這種形式。這種做法只有純附加的掃描能用:一個要動到多個物件的工作會把中間狀態暴露給讀取者,工作崩潰就留下一張壞掉的表,而 S3 的 LIST 又是最終一致的,連寫入者自己剛放上去的物件都可能看不到。另一條路是 Snowflake 或 Hive ACID 這類封閉式引擎,把資料表的真實內容交給自己那套強一致的 metadata 服務管理,代價是必須常駐一個高可用服務、分割數到百萬級就成為瓶頸、每個引擎都要額外投入接連接器,而且被綁在單一供應商上。與此同時,GDPR 與單純的資料修補需求,正把全表更新硬塞給當初按不可變設計的資料集。作者提到,Databricks 雲端服務最初幾年,大約有一半的技術支援升級案件,都是雲端儲存策略造成的資料毀損、一致性或效能問題。

問題 — 當時真正壞掉的地方

  • 雲端物件儲存不提供跨 key 的原子性,因此一個要改寫多個 Parquet 物件的查詢,會讓並行讀取者看到寫到一半的狀態,崩潰時更會直接留下毀損的資料表。
  • 主流物件儲存對單一 key 只有最終一致性、跨 key 完全沒有保證,客戶端可能只看到交易新物件中的一部分;S3 的 LIST 甚至可能回傳不到剛剛 PUT 上去的物件。
  • 規模一大,metadata 操作就變得昂貴:S3 的 LIST 每次最多回傳 1000 個 key,每次呼叫要數十到數百毫秒,用循序方式列出數百萬個物件的資料集要花上好幾分鐘。
  • 每個物件的 min/max 統計藏在 Parquet footer 裡,做 data skipping 等於每個物件都要多一次高延遲讀取;在物件儲存上,這些跳過檢查本身可能比查詢還久。
  • 物件儲存完全沒有資料倉儲該有的管理功能:沒有資料表版本、沒有辦法回復一個崩潰的更新工作,也沒有記錄誰改了什麼的稽核日誌。
  • 用一套獨立強一致 metadata 服務來解決上述問題的封閉式儲存引擎,必須常駐一個高可用服務、所有 I/O 都得經過它,替 Spark、TensorFlow、PyTorch 等引擎寫連接器更費工,還會把使用者鎖在單一供應商上。

核心概念 — 主要貢獻,以及它們為何成立

交易日誌就放在物件儲存裡

Delta Lake 最關鍵的一步,是把「這張表由哪些物件組成」記在一份與資料同一個 bucket、放在 _delta_log 前綴底下的 WAL,而不是交給外部 catalog。真正的事實來源是這份日誌,不是對目錄做列舉,這正是多物件變更得以原子化的原因:新增、移除或替換任意數量的 Parquet 物件,都只是一筆新的日誌記錄。因為日誌本身也只是物件,就不需要任何伺服器常駐持有資料表狀態,運算資源可以只在查詢時才開,而資料表的可用性恰好等同底層物件儲存的可用性。光是這一點,就同時把 Delta Lake 與「一堆檔案」的做法、以及依賴 metadata 服務的封閉式引擎區隔開來。

開放的 Parquet 資料,極小的連接器面積

資料表內容仍然是一般的 Apache Parquet 物件,可依 Hive 風格的分割目錄擺放,每個物件由寫入者取一個 GUID 當名字。作者選 Parquet 是因為它是欄式、壓縮方式多樣、支援半結構化資料的巢狀型別,而且在許多引擎裡都已有高效實作;ORC 大致也行得通,只是 Parquet 在 Spark 上的支援最成熟。結果是任何讀得懂 Parquet 的引擎,只需要一個小連接器去查出該讀哪些物件,而不必移植整套儲存引擎。對連這都嫌重的引擎,Delta Lake 還會產生 symlink manifest 檔(原本是 Hive 為了符號連結而加的機制),讓 Presto、Athena、Redshift 與 Snowflake 能以外部表的形式讀到一致的快照。

checkpoint 把 metadata 變成欄式資料表

從頭重播每一筆 JSON 日誌記錄會和 LIST 一樣慢,因此客戶端會定期把日誌壓縮合併成 Parquet 格式的 checkpoint,預設每 10 筆交易一次。checkpoint 會丟掉可證明為多餘的 action:被後續 remove 抵銷掉的 add、被較新 add 取代的舊 add、同一個 appId 中被取代的舊 txn,以及過期的 metaData 與 protocol。剩下的就是一個欄式檔案,表中每個仍存活的物件各有一筆 add,附帶該物件的筆數、每欄 min/max 與 null 數。於是「找出這個選擇性查詢該讀哪些物件」變成對一個 Parquet 檔的向量化掃描;作者表示,這幾乎總是比在物件儲存上做 LIST 再逐一讀 footer 來得快。

以單一原子寫入撐起樂觀並行控制

每一筆寫入交易最後都收斂成一個原子步驟:在讀過版本 r 之後,建立編號 r+1 的日誌記錄,若已被別人建立就失敗。資料物件是先平行寫好的,名字是 GUID,在提交之前沒有任何東西指向它們,所以提交前崩潰只會留下孤兒物件,永遠不會讓人看到寫到一半的資料表。輸掉競爭的寫入者可以直接重試,並依查詢語意決定是否沿用已經寫好的資料物件,用更大的編號再提交一次。這是教科書式的樂觀並行控制,但它在物件儲存上的全部實作成本只是一次 put-if-absent,這也是為什麼這套設計不需要鎖管理員,在多數雲端上更完全不需要協調者。

不可變性換來 time travel 與快取

資料物件與日誌記錄都不會原地修改,移除則以帶時間戳的 tombstone 記在日誌裡,實體刪除會延後到超過每張表設定的保留期限之後。因此讀取舊版本不過是在較舊的日誌記錄編號上重建狀態,對外以 SQL 的 AS OF timestamp 與 VERSION AS OF commit_id 呈現;使用者可以拿資料表自己的過去做 MERGE 來回復一次失敗的 pipeline,MLflow 也能自動記下某個模型究竟是用哪個資料表版本訓練的。同樣的不可變性讓本地快取變得安全:既然沒有物件的內容會在快取副本底下改變,Databricks 就能把資料與日誌物件一起快取在叢集的 SSD 上,完全不需要失效協定。兩個在可變儲存上各自都要一大套機制的功能,就這樣從同一個設計選擇裡掉出來。

可交易化的資料佈局最佳化

因為「資料表對應到哪些物件」這件事是交易化的,背景程序可以在查詢照跑的同時重寫實體佈局。OPTIMIZE 會把小物件壓縮合併到預設目標大小 1 GB,並補齊缺少的統計;在 Databricks 服務上,AUTO OPTIMIZE 則對新寫入的資料自動做這件事。ZORDER BY 沿著 Morton 空間填充曲線在多個欄位上重排記錄,讓每個物件在每個指定維度上都只涵蓋很窄的值域,而不是只有單一維度窄;對「不同時候用不同欄位過濾」的工作負載,min/max skipping 的效果因此倍增。關鍵在於這些重寫會把 dataChange 設為 false,等於告訴串流消費端這筆記錄沒有搬動任何資料,可以整筆略過。

串流、data lake 與資料倉儲共用一套儲存

寫入壓縮合併、exactly-once 的 txn 記錄與廉價的日誌 tailing 合起來,讓一張 Delta 表同時能當訊息佇列用:生產者以低延遲提交小物件,消費者從自己處理過的最後一個日誌編號往後 LIST,背景工作稍後再把小物件併大,兩邊都不受干擾。再加上 ACID 更新、以統計驅動的 skipping 與 SSD 快取,一張物件儲存上的資料表就能同時扮演過去需要訊息匯流排、data lake 與資料倉儲三套系統、三份資料副本才做得到的 ETL、串流與 BI 角色。論文把這個結果稱為 lakehouse:把標準的 DBMS 管理功能直接套用在低成本的物件儲存上。作者表示,許多客戶因此把多系統的 pipeline 收斂成幾張 Delta 表,同時省下儲存成本與維運負擔。

運作方式 — 具體的機制

資料表在物件儲存上的佈局

一張 Delta 表就是一個目錄,或者說一組共用同一段 key 前綴的物件,裡面放著 Parquet 資料物件,外加一個 _delta_log 子目錄。資料物件可以放在 date=2020-01-01/ 這類 Hive 風格的分割子目錄下,每個物件由寫入者產生一個 GUID 當名字。_delta_log 裡則是以補零遞增整數命名的日誌記錄(000001.json、000002.json 等)、偶爾出現的 checkpoint(例如 000003.parquet),以及一個內容為最新 checkpoint 編號的 _last_checkpoint 檔。補零是刻意的:它讓物件儲存原本就有的字典序 LIST 表現得像數值範圍掃描,客戶端因此能直接要到某個編號之後的全部記錄。

日誌記錄的 action 語彙

每個 .json 日誌記錄裝著一組 action,套用在前一個資料表版本上。metaData 會整個覆寫資料表的 schema、分割欄位名稱、資料檔格式與 append-only 之類的設定,且必須出現在資料表的第一個版本裡。add 指名一個資料物件,可附上它的筆數與每欄 min/max、null 數;同一路徑後來的 add 會取代舊的統計,這正是舊表升級到更豐富統計的方式。remove 帶有時間戳,並以 tombstone 形式留在日誌與 checkpoint 中,直到過了保留期限、物件被實體刪除為止,這也讓並行讀取者能繼續在較舊的快照上執行。protocol 用來提高讀寫這張表所需的 Delta 協定版本,commitInfo 記錄來源資訊(例如是哪個使用者執行的操作),txn 則替應用程式保存一組 (appId, version)。

checkpoint 的產生與尋找

任何客戶端都可以替日誌寫出涵蓋到某個記錄編號的 checkpoint,例如替 000003.json(含)以前的記錄寫出 000003.parquet;Databricks 的客戶端預設每 10 筆交易做一次。checkpoint 保留表中每個仍存活物件的一筆 add、尚未過保留期的 remove tombstone,以及合併後的 txn、protocol 與 metaData 狀態。寫完之後,若編號比檔案裡原本的還新,客戶端才更新 _last_checkpoint。這一步純粹是效能最佳化:客戶端在寫 checkpoint 途中掛掉,或寫完 Parquet 卻沒更新 _last_checkpoint,都不會造成任何毀損,讀取者只是退回較舊的 checkpoint、再多讀一段日誌尾端而已。

最終一致性下的讀取協定

讀取者的五個步驟是:(1) 讀 _last_checkpoint 取得一個近期的 checkpoint 編號;(2) 以該編號(沒有就用 0)為起點做 LIST,找出更新的 .json 與 .parquet;(3) 重播 checkpoint 加上這些日誌記錄,算出「有 add 但沒有對應 remove」的物件集合與它們的統計;(4) 用統計把範圍縮到查詢真正需要的物件;(5) 在叢集上平行讀取這些物件。每一步都是照「資料可能過時」寫的:_last_checkpoint 過時只會讓 LIST 多花一點;LIST 回傳了 000004.json 與 000006.json 卻少了 000005.json,客戶端仍以最大編號作為目標版本,等待缺口變得可見;工作節點暫時看不到日誌中提到的資料物件,就等一下再重試。狀態重建本身是平行的,Spark 連接器就是用 Spark job 去讀 checkpoint Parquet 與日誌物件。

寫入協定與提交點

寫入者的五個步驟是:(1) 用讀取協定的前兩步找出近期的日誌記錄編號 r;(2) 若交易需要,就讀取版本 r 的資料;(3) 平行地以新的 GUID 名字,把新資料物件寫進正確的資料目錄;(4) 嘗試建立 r+1 的 .json 日誌物件,且僅在沒有其他客戶端寫過它時成立;(5) 選擇性地寫 checkpoint,再更新 _last_checkpoint。第 4 步就是這筆交易本身:它是唯一必須原子的步驟,也是寫入之所以可序列化的原因。第 3 步寫出的東西在第 4 步成功前對任何讀取者都不可見,因為讀取者只透過日誌認識物件,所以在更早任何時點中止,代價都只是幾個孤兒物件。若第 4 步因為別人先搶到 r+1 而失敗,交易可以重試,並依查詢語意決定是否沿用已經寫好的資料物件。

在各種儲存系統上做到原子建立日誌記錄

提交需要 put-if-absent,但不是每個大規模儲存系統都有,因此第 4 步在不同後端有不同實作。Google Cloud Storage 與 Azure Blob Store 直接提供原子的 put-if-absent,就直接沿用。在 HDFS 這類分散式檔案系統以及 Azure Data Lake Storage 上,改用原子 rename:把暫存檔改名為 000004.json,目標已存在就失敗。Amazon S3 兩者都沒有,所以 Databricks 的部署另外跑一個輕量的協調服務,確保同一個日誌編號只有一個客戶端寫得成;它只擋在日誌寫入路徑上,不參與讀取、也不參與資料操作,因此負載很小。開源的 Spark 連接器則改用 driver 的記憶體內狀態發放不同的日誌編號,讓單一 Spark 叢集內的並行操作仍然正確,另外提供可插拔的 LogStore 類別,讓使用者換上自己的強一致協調機制。

隔離等級與交易速率上限

由於每個日誌記錄編號只可能被一筆交易佔走,所有寫入交易都是可序列化的,而其序列順序就是日誌記錄編號遞增的順序。照讀取協定跑的讀取者得到 snapshot isolation;需要可序列化讀取的客戶端,可以跑一筆帶 dummy write 的讀寫交易。連接器另外會在記憶體裡快取每張表看過的最大日誌編號,因此即使在 snapshot isolation 下,客戶端也能讀到自己的寫入,並看到單調遞增的版本序列。真正的天花板是那次原子寫入的延遲,數十到數百毫秒,把提交速率壓在每秒數筆;一如所有樂觀機制,更高的請求速率只會轉成提交失敗。snapshot isolation 的讀取則完全不競爭,可以任意擴展。交易範圍限於單一資料表,因為每張表各有自己的日誌。

論文證明了什麼 — 量測數據與證明

  • 在 16 節點的 i3.2xlarge AWS 叢集上,查詢一張只有 33,000,000 列、但分割數很多且存在 S3 上的小表:託管 Hive 在 10,000 個分割時就超過一小時,託管 Presto 在 100,000 個分割時超過一小時;Databricks Runtime 直接列出 Parquet 檔案在 100,000 個分割時要 450 秒;Delta Lake 在 1,000,000 個分割時只要 108 秒,日誌快取在 SSD 上更只要 17 秒。
  • 在一張 100 個物件的合成網路流量表上(32 位元 IP 與 16 位元 port 均勻隨機產生),依 (sourceIP, sourcePort, destIP, destPort) 做全域排序時,過濾 sourceIP 可跳過 99% 的物件,其餘三個欄位則是 0%,平均 25%;改用同樣四欄的 Z-order 之後,每個維度至少可跳過 43%,平均 54%。
  • 論文所述那家資安客戶的一份真實 500 TB 網路流量資料集,以類似欄位做 Z-order 之後,多屬性查詢可跳過表中 93% 的資料。
  • TPC-DS power test 在 S3 上跑 1 TB 資料、一台 master 加八台 i3.2xlarge worker、取三次平均:Databricks 搭 Delta 為 0.93 小時、Databricks 搭 Parquet 為 0.99、第三方 Spark 搭 Parquet 為 1.44、第三方 Presto 搭 Parquet 為 3.76。
  • 在一台 master 加八台 i3.2xlarge worker 上,把 400 GB 的 TPC-DS store_sales 從 CSV 載入,寫成 Delta 與寫成 Parquet 所花時間相近,顯示逐物件收集統計相對於其餘載入工作並未帶來明顯額外開銷。
  • 在生產環境中,Delta Lake 運行於數千家 Databricks 客戶、每天處理 EB 等級資料,約佔該服務整體工作負載的一半,規模最大的部署管理著數十億個物件;作者表示與雲端儲存相關的技術支援案件,從約占一半降到幾乎為零,在極高維度的資料集上最多有 100 倍的加速。

限制與取捨 — 論文自承的,以及後來被發現的

  • 論文自承交易只在單一資料表內可序列化,因為每張表各有自己的交易日誌;讓多張表共用同一份日誌可以解除這個限制,但會讓那唯一的附加點競爭變高。
  • 論文自承寫入交易速率受限於物件儲存 put 的數十到數百毫秒延遲,只有每秒數筆,而且樂觀並行控制會把超出的請求速率直接變成提交失敗;作者的論證是實務上寫入多為大批次,因此可以接受。
  • 論文自承串流延遲受制於底層物件儲存,毫秒級串流做不到,實際目標是數秒等級;也因此 Delta Lake 只能在容忍這個延遲的場景取代訊息匯流排。
  • 論文自承唯一的索引就是每個物件的 min/max 統計,Bloom filter 索引當時只停在原型階段;高度選擇性的點查詢因此仍取決於 Z-order 佈局做得好不好。
  • 「不需要任何伺服器」這個說法在最常見的 S3 上是有但書的:Databricks 在那裡另外跑一個提交協調服務,開源連接器則只能在單一 Spark driver 內把提交序列化。後續發展緩解了這一點——S3 在 2020 年底提供強一致的 read-after-write,2024 年再加上條件式寫入原語,而整個生態實務上也還是收斂到 catalog 服務。

它後來變成什麼 — 繼承這個想法的系統

Delta Lake 於 2019 年以 Apache 2 授權開源,與 Apache Hudi、Apache Iceberg 一起把「表格式」確立為技術堆疊中獨立的一層:一份疊在 Parquet 之上、由 add 與 remove action 組成的開放日誌,多個引擎都讀得到,卻不歸任何一家所有。論文的 lakehouse 說法成為業界描述分析架構的主流敘事,同一批作者在 CIDR 2021 的 lakehouse 論文中進一步展開,也被 Databricks、Snowflake、Google BigQuery 與 AWS 拿來當作產品定位。Iceberg 採用同樣「日誌放在物件儲存裡」的想法但用不同的 manifest 結構,取得了廣泛的中立採用,以致 Databricks 在 2024 年買下 Tabular,並推出 Delta UniForm,讓同一份 Parquet 物件能同時呈現 Delta 與 Iceberg 兩種 metadata。個別機制也活得比出身更久:AS OF 語法的 time travel、在 lake 表上做 MERGE、以 dataChange 為代表的增量日誌 tailing 串流來源,以及用帶統計的 manifest 取代 LIST,如今在 Trino、Flink、DuckDB、Spark 與 Apache Paimon 都被視為理所當然的功能。論文中那個 S3 協調服務的權宜之計則大致退場,先是 S3 在 2020 年 12 月改為強一致的 read-after-write,再來是 2024 年的條件式寫入;而 Unity Catalog 與 Iceberg REST catalog 這類 catalog 服務,又把協調者重新請了回來,正是為了做到論文列為未來工作的跨表交易。

論文原文 — 逐字引用

“The core idea of Delta Lake is simple: we maintain information about which objects are part of a Delta table in an ACID manner, using a write-ahead log that is itself stored in the cloud object store.”

§1

“However, Delta Lake takes 108 seconds even with 1 million partitions, and only 17 seconds if the log is cached on SSDs.”

§6.1

“Delta Lake is implemented solely as a storage format and a set of access protocols for clients, making it simple to operate and highly available, and giving clients direct, high-bandwidth access to the object store.”

§9

術語 — 依本篇論文的用法

Delta table
物件儲存上的一個目錄,或者說一組共用同一段 key 前綴的物件,裡面放著 Parquet 資料物件與一個 _delta_log 子目錄。定義資料表包含哪些物件的是日誌,而不是對目錄做列舉。
交易日誌(_delta_log)
一連串以補零整數命名的 JSON 記錄,每筆內容是套用在前一個資料表版本上的 action 陣列;它是這張表的 WAL,也是唯一的事實來源。
Checkpoint
把日誌壓縮合併到某個記錄編號、並移除多餘 action 之後寫成的 Parquet 檔,預設每 10 筆交易產生一次,讓讀取者不必從頭重播整份日誌。
add 與 remove action
在資料表上掛上或拆下單一資料物件的日誌 action。add 可帶該物件的筆數與每欄 min/max、null 數;remove 則是帶時間戳的 tombstone,會保留到過了保留期限為止。
Data skipping 統計
存在日誌裡(而非 Parquet footer 裡)的每物件 min/max、null 數與筆數,讓查詢規劃器用一次對 checkpoint 的欄式掃描就淘汰掉不相關物件,不必每個物件各做一次高延遲讀取。
dataChange 旗標
add 與 remove action 上的布林欄位;設為 false 表示這筆提交只是重排既有資料或補上統計,串流消費端因此可以略過壓縮合併與 Z-order 重寫。
txn action
由應用程式提供的一組 (appId, version),與資料變更放在同一筆日誌記錄中原子提交;Structured Streaming 用它讓寫入具備冪等性,達成 exactly-once 語意。
Z-ordering
沿著 Morton 空間填充曲線在多個欄位上重排記錄,使每個物件在每個所選維度上都只涵蓋很窄的值域,讓 min/max skipping 對多屬性過濾的效果倍增。
Lakehouse
論文替這套結果取的名字:把交易、版本管理與稽核日誌等標準 DBMS 管理功能,直接套用在低成本雲端物件儲存上的資料表。

在時間軸上的位置 — 這篇論文在整段故事中的座標

在時間軸上查看