跳至主要內容
論文精煉 · Query processing

Presto: SQL on Everything

一套具適應性的分散式 SQL 引擎,同時支撐次秒級儀表板與長達數小時的 ETL,並可插接數十種資料來源。

作者Raghav Sethi、Martin Traverso、Dain Sundstrom、David Phillips 等人(Facebook, Inc.) 發表於ICDE 2019 年份2012–2013
閱讀原始論文 PDF 所有論文

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

Presto 是 Facebook 打造的開源分散式 SQL 查詢引擎,目的是取代組織為了不同類型分析而被迫並存的多套互不相容的類 SQL 系統。單一 coordinator 負責解析、規劃與查詢最佳化,再把 stage 化為 task 派送到 worker;worker 共用一個長生命週期的 JVM,以管線化的 in-memory driver loop 在欄式 page 上處理資料。SQL 之下的一切都藏在 Connector API 後面,這個介面分成 Metadata、Data Location、Data Source 與 Data Sink 四部分,因此同一套引擎可以讀 Hive 資料倉儲、分片 MySQL、Raptor 快閃儲存、鍵值系統與 Kafka,甚至在同一個查詢裡同時讀。Presto 刻意不用大量設定參數,而是選擇適應性:延遲式 split 指派、以 HTTP long-polling 實作並具端到端 backpressure 的 shuffle、CPU 的 multi-level feedback queue、可超額配置的記憶體 pool、動態寫入並行度,以及對 dictionary 編碼區塊的推測式處理。論文指出,這一套引擎既服務 50ms 的廣告主儀表板,也跑五小時的 ETL,每天處理數百 PB 資料與數千兆(quadrillion)列。

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

2013 年前後,Facebook 的分析工作跑在 Hive 上:類 SQL 查詢被編譯成 MapReduce 或 Tez 工作,stage 之間的中間結果落到檔案系統。這種設計換來了容錯,卻讓再簡單的查詢也以「分鐘」為時間下限,互動式探索、BI 工具與面向使用者的儀表板在資料倉儲上等於不可行。另一條路是 Vertica、Teradata、Redshift 或 Exadata 這類 MPP 資料倉儲,但它們只有在資料被載入自家儲存後才有巔峰效能,代價是搬運數 PB 資料並維護第二份真實來源。Spark SQL 與 Impala 縮短了差距,但前者仍缺乏端到端管線化,後者則侷限在 Hadoop 生態系內。實務結果是:同一個組織按延遲級距養了好幾套互不相容的類 SQL 系統,使用者還得自己判斷該把查詢丟給哪一套。

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

  • 當組織為了涵蓋不同類型的分析問題,被迫部署多套互不相容的類 SQL 系統時,易用性就崩壞了。
  • 同一套引擎必須橫跨相差五個數量級的延遲需求:從 50ms 的廣告主儀表板,到輸入超過 100TB、跑五小時的 ETL 工作。
  • 像 Hive 與 Spark SQL 這類在 stage 之間把 shuffle 資料寫入檔案系統的系統雖然換得容錯,卻增加了延遲,不適合互動式場景。
  • 傳統資料倉儲產品讀取外部資料的能力有限,只有在資料先載入自家內部儲存後才最快,因而強迫使用者付出昂貴的匯入成本。
  • 每個 worker 用單一長生命週期的共用 JVM 承載數百個並行查詢,必須自建整合式的 CPU 排程、記憶體隔離與准入控制——這些在每查詢一個行程的模型裡是免費的。
  • 若把每節點與全域記憶體上限設得保守到足以吸收 skew,500 節點叢集上能並行執行的查詢數會少到不可接受;而 Hive connector 單一查詢就可能列舉出數百萬個 split,光是這些 metadata 就會在查詢開始執行前耗盡 coordinator 的記憶體。

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

一套引擎,四種工作負載

論文以 Facebook 的四個生產使用情境來鋪陳 Presto:Interactive Analytics、Batch ETL、A/B Testing 與 Developer/Advertiser Analytics,它們幾乎在每個面向上都不同。Interactive Analytics 在約 50GB 到 3TB 的壓縮資料上跑 50 至 100 個並行探索式查詢,使用者盯著時鐘等;Batch ETL 願意犧牲延遲換取資源效率與叢集吞吐量;A/B Testing 需要即時算出完整且正確的結果,並在 5 到 30 秒內讓使用者任意切片;Developer/Advertiser Analytics 則要求 50ms 到 5 秒的回應與 99.999% 可用性。論文主張的不是同一組設定能服務四者,而是同一份程式碼、同一種 SQL 方言與同一套 Connector API 能被組態成這四種樣貌。這正是「查詢引擎」與「產品家族」的分水嶺。

Connector API 作為聯邦化契約

Presto 沒有把外部系統當成退而求其次的路徑,而是把每個資料來源都放在一個插件介面之後,並將它拆成 Metadata API、Data Location API、Data Source API 與 Data Sink API。這樣拆分之所以重要,是因為它把最佳化器在規劃期需要的東西(schema、統計資訊、實體 layout、分割、排序、索引)與 worker 在執行期需要的東西(要讀的 split、要回傳的 page、要寫入的 sink)分了開來。由於 connector 回報的是 layout 而不只是資料列,最佳化器才能把述詞下推到分片 MySQL、對生產環境的鍵值儲存選用 index nested-loop join,或在 join 兩側已依 join 鍵分割時直接消掉一次 shuffle。主儲存庫已有十多個社群貢獻的 connector,而通用的 Thrift RPC connector 讓「替專有服務加上 SQL 介面」簡化成實作約半打 RPC 端點。

適應性優於可組態性

論文把這點明講成一條工程哲學:一個執行任意使用者自訂運算的多租戶引擎,必須適應的不只是不同的查詢特性,還有這些特性的各種組合,因此逐一為工作負載調參的做法根本擴展不了。具體而言,split 是延遲指派給佇列最短的 task,因此慢 split 與慢節點會自我修正;輸出緩衝區使用率會節流有效並行度;輸入緩衝區的監測決定 HTTP 請求並行度;當上游 stage 塞住時寫入並行度會自動增加;page processor 還會推測該處理整個 dictionary 還是只處理被引用到的索引。在端到端 backpressure 出現之前,少數幾個客戶端很慢的工作就能佔住數十 GB 緩衝記憶體,並拖垮同時執行的延遲敏感查詢。適應性正是讓單一叢集能承載沒人事先列舉過的查詢形狀的關鍵。

以 HTTP 承載的管線化 in-memory shuffle

Presto 透過 in-memory 緩衝區以 HTTP long-polling 交換中間結果:產出端的 task 把輸出留在緩衝區,消費端輪詢,而每次回應中夾帶的 token 就隱含地確認了前一段資料,因此不需要額外的 ack 通道。stage 之間完全不落磁碟,資料一產生就往下游串流,這也是某些查詢形狀能在讀完全部輸入前就回傳結果的原因。這是 Presto 與 Hive、Spark SQL 最根本的分歧,而且是刻意的取捨:放棄物化 shuffle 天然提供的重試點,換取互動式與面向使用者場景所需的延遲。long-polling 尤其能在傳輸量很小時壓低回應時間,而儀表板查詢正是這種情況。

細粒度的整合式資源管理

由於每個 worker 上多個查詢共用一個 JVM,Presto 自帶排程器,而不是仰賴作業系統或叢集管理器。CPU 採用協作式多工,每個 split 單次最多佔用執行緒一秒鐘的 quanta;task 則依累積 CPU 時間被歸入 multi-level feedback queue 的五個層級,因此便宜的查詢留在高層快速結束,昂貴的查詢逐步下沉。記憶體則區分為 user memory(使用者能從查詢與資料推理出來的用量)與 system memory(實作副產物,例如 shuffle 緩衝區),兩者各有每節點與全域上限。引擎刻意超額配置叢集記憶體,賭的是不會所有查詢在同一時刻於同一節點都用到上限,並以 spilling 與 reserved pool 作為這個賭注的後盾。

為 JIT 而生的程式碼產生

Presto 對運算式求值與整個運算子都產生 JVM bytecode,而有趣的論點在於:為什麼在這個平台上通用直譯必然失敗。因為引擎每個 quanta 就會在不相干 task 的 split 之間切換,共用的緊湊處理迴圈所收集到的 profiling 資訊會被其他查詢污染,JIT 因此永遠無法特化它。為每個 task 產生獨立的 Java 類別,讓每個 task 擁有自己的 profile,JIT 便能將單型(monomorphic)呼叫點內聯、對具型別的欄位展開迴圈、自動向量化,並隨著 task 在生命週期中看到的資料改變而重新最佳化。引擎也順著 G1 收集器的脾氣設計——避免配置超過 humongous 門檻的物件、以分段與扁平陣列取代龐大的相互參照物件圖——因為否則 GC 行為就會直接決定吞吐量。

運作方式 — 具體的機制

Coordinator 上的查詢生命週期

客戶端以 RESTful HTTP 介面送出 SQL,coordinator 先評估排隊政策,接著由 ANTLR 為基礎的 parser 產生語法樹,analyzer 用它決定型別與強制轉型、解析函式與作用域,並抽出子查詢、聚合與 window function。邏輯 planner 產出的是純邏輯的 plan node 樹,不帶任何執行資訊,每個節點的子節點就是它的輸入。最佳化器接著貪婪地套用轉換規則直到達到不動點:每條規則有一個能比對 plan 子樹的 pattern,比對成功就改寫成邏輯等價的子計畫。Presto 內建述詞與 limit 下推、欄位裁剪與去相關(decorrelation),另有兩項會使用表格與欄位統計資訊的成本式最佳化——join 策略選擇與 join 重排——而更完整的 Cascades 式搜尋當時仍在進行中。

以減少 shuffle 為目標的分散式規劃

最佳化器把計畫切成 stage,也就是能跨 worker 平行執行的單位,並在其間插入具緩衝的 in-memory shuffle;由於 shuffle 會付出延遲、緩衝記憶體與 CPU 成本,把 shuffle 數量壓到最低是實體規劃的核心目標。plan node 會宣告輸出屬性——分割、排序、bucketing、grouping——也能宣告 required 與 preferred 屬性,於是多餘的 shuffle 直接被消去,其餘的則調整參數。Presto 貪婪地挑選能滿足最多 required 屬性的分割方式,這可能表示改用較少的欄位分割並接受更嚴重的 skew。當 connector 提供 layout 時,效果非常實際:A/B Testing 中 join 兩側本來就依同一欄位分割,因此可用 co-located join 省掉整次 shuffle;而圖 3 那個需要四次 shuffle 的天真計畫,最後會塌縮成單一個資料處理 stage。

stage、task 與 split 的排程

stage 順序由兩種政策決定。all-at-once 同時排程所有 stage 以最小化 wall clock 時間,服務延遲敏感的使用情境;phased execution 則找出資料流圖中必須同時啟動才能避免死結的強連通分量,並依拓撲順序執行,因此 hash join 在 build 端完成前不會排程 probe 端的串流,對批次工作的記憶體效率好得多。task 排程再把 stage 分成 leaf 與 intermediate:leaf stage 從 connector 讀資料,在沒有額外限制且 split 夠多時會排到叢集中每一個 worker 上,因為解壓縮、解碼與過濾佔了大部分 CPU 且高度可平行化;排程器也能依插件提供的網路拓撲,優先選擇 rack-local 讀取。intermediate stage 的 task 可以放在任何節點,而且引擎可以在執行期間動態改變其數量。

延遲式 split 列舉與指派

split 是外部儲存系統中某塊可定址資料的不透明控制代碼,內容由 connector 定義——分散式檔案系統是檔案路徑加位移量,Redis 則是表格資訊加上鍵與值的格式以及主機清單。Presto 要求 connector 分成小批次列舉 split,並延遲指派給 task,換來四項好處:查詢回應時間與列舉時間解耦,而 Hive connector 光是列出分割與其中的檔案就可能花上數分鐘;帶 LIMIT 或被提早取消的查詢常常在列舉完成前就結束;coordinator 把新的 split 指派給佇列最短的 task,吸收了各 split CPU 成本與各 worker 速度的差異;coordinator 記憶體也不必一次容納數百萬個 split 的 metadata。每個 leaf stage 的 task 至少要拿到一個 split 才具備執行資格,而 intermediate task 永遠可執行,只有被中止或所有上游 task 完成時才結束。論文明講的代價是:準確回報查詢進度變得困難。

本機執行、緩衝區與 backpressure

在 task 內部,driver loop 在能夠推進的運算子配對之間搬動 page——page 是資料列序列的欄式編碼,由每欄一個且採扁平記憶體表示的 Block 組成——而不是採用 Volcano 式遞迴迭代器的 pull 模型。之所以選這個迴圈,正是因為運算子可以被帶到已知狀態後讓出執行緒,而不是無限期阻塞,這才讓協作式多工成立;此外,當輸出緩衝區滿、輸入緩衝區空或記憶體耗盡時,排程器也會提早切換。在網路端,引擎監看輸出緩衝區使用率,當它持續偏高就減少可執行的 split 以降低有效並行度,這既讓網路資源分配更公平,也避免一個下載 10 到 50MB 的緩慢 BI 客戶端佔住數十 GB 緩衝區。接收端則追蹤每次請求傳輸量的移動平均,據此計算目標 HTTP 並行度,產生的 backpressure 會往上游傳遞。寫入則反向套用同一個想法:當產出資料的 stage 超過緩衝區使用率門檻時,就在更多節點上新增 task 以提高寫入並行度,在 S3 小檔爆量與寫入吞吐量之間取得平衡。

記憶體 pool、spilling 與 reserved pool

所有非瑣碎的配置都必須歸類為 user 或 system memory,並向對應的 pool 預約;超過全域或每節點上限的查詢會被終止,而當節點記憶體耗盡時,引擎會停住 task 處理來阻擋新的預約。論文把算式攤開來:500 節點叢集、每節點 100GB 查詢記憶體、單一查詢全域上限 5TB,理論上可容納十個查詢;但若要允許 2:1 的 skew,每節點上限就得壓到 20GB,保證能並行的只剩五個——少到不可接受,因此才刻意超額配置。有兩個機制維持超額配置叢集的健康。spilling 會依執行時間由小到大對合格的 task 進行記憶體撤回,直到最後一次請求能被滿足為止,撤回方式是把 hash join 與聚合的狀態寫到磁碟。若未設定 spilling 或已無可撤回的記憶體,就換 reserved pool 出手:記憶體吃緊節點上用量最大的查詢會被提升到所有 worker 的 reserved pool,而為了避免不同 worker 各自卡住不同查詢造成死結,全叢集同一時間只允許一個查詢進入 reserved pool。

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

  • 在 100 節點叢集上(每節點 28 核心 Xeon E5-2680 v4、2.40GHz、1.6TB 快閃儲存、256GB DDR4)執行 scale factor 30TB 的 TPC-DS 低記憶體子集,同一批查詢分別以 Raptor、無統計資訊的 Hive/HDFS、以及帶表格與欄位統計資訊的 Hive/HDFS 三種設定執行;圖 6 顯示在查詢與叢集組態都不變的情況下,執行時間受 connector 特性影響極大。
  • 以生產環境 Batch ETL 工作負載樣本量測,lazy data loading 使抓取的資料量減少 78%、載入的 cell 數減少 22%、總 CPU 時間減少 14%。
  • 圖 7 針對四種生產使用情境的執行時間 CDF,在對數座標上橫跨約 16ms 到約 19 小時,證明同一套引擎既能服務 20 到 100ms 需求的 Web 場景,也能承載程式排程、跑上數小時的 ETL 工作。
  • 某 Interactive Analytics 叢集的四小時追蹤(圖 8)顯示,即使並行度從尖峰的 44 個查詢降到低點的 8 個,Presto 仍維持 worker 平均約 90% 的 CPU 使用率,並在查詢獲准後數毫秒內就配給它相當大比例的叢集 CPU。
  • 表 I 記錄了實際部署的操作點:Developer/Advertiser Analytics 為 50ms 至 5 秒,數十個節點、數百個並行查詢,走分片 MySQL;A/B Testing 為 1 至 25 秒,數百個節點,走 Raptor;Interactive Analytics 為 10 秒至 30 分鐘,數百個節點、50 至 100 個並行查詢,走 Hive/HDFS;Batch ETL 為 20 分鐘至 5 小時,最多 1000 個節點,走 Hive/HDFS。
  • 2018 年底(版本 0.211)回報的營運規模:叢集規模最大約 1000 個節點,每天處理數百 PB 資料與數千兆列,中位數的 worker 節點輸出約 10,000 個即時效能計數器。

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

  • 論文自承:截至 2018 年底,Presto 對 coordinator 或 worker 當機沒有任何有意義的內建容錯——coordinator 故障會讓整個叢集不可用,worker 當機會讓該節點上所有查詢失敗,復原完全交給客戶端重試、備援 coordinator 與多個同時上線的叢集。
  • 論文自承:spilling 雖已支援 hash join 與聚合,但 Facebook 沒有任何部署設定啟用它,因為叢集本身就有數 TB 的分散式記憶體、使用者偏好全記憶體執行帶來的可預測延遲,而在 shared-storage 部署中加本機磁碟會提高硬體成本。
  • 論文自承:reserved pool 相當浪費,因為每個節點上的 reserved pool 都必須大到足以容納一個逼近本機記憶體上限的查詢,全叢集同時只有一個查詢能佔用它,而 general pool 已耗盡節點上的其他 task 只能停等該查詢結束。
  • 論文自承:延遲式 split 列舉讓查詢進度難以準確估算與回報;而貪婪的 shuffle 削減啟發式可能改用較少欄位分割,反而加重分割 skew。
  • 後續發展揭露:論文發表時成本式最佳化仍不完整——只有 join 重排與 join 策略選擇會用到統計資訊,Cascades 式搜尋還在進行中——而兩條後繼專案日後都補上了本設計刻意省略的 dynamic filtering 與以物化 exchange 實現的容錯執行。

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

Presto 成為「聯邦式、與儲存無關的 SQL 引擎」的樣板:Amazon Athena 直接建構在它之上,Uber、Netflix、Airbnb、Bloomberg 與 LinkedIn 都在生產環境使用,Qubole、Treasure Data 與 Starburst 則圍繞它推出商業產品。2019 年專案分裂,原作者的分支先成為 PrestoSQL 再更名為 Trino,Facebook 這一脈則以 PrestoDB 之名納入 Linux Foundation 的 Presto Foundation。Trino 後來逐一補上本論文點名的缺口——完整的成本式最佳化器、以 build 端值裁剪 probe 端掃描的 dynamic filtering,以及把 exchange 物化到物件儲存的容錯執行,而後者正是論文說自己正在評估的可選 checkpointing。Meta 則往另一個方向推進,用 C++ 重寫 worker 的執行層,成果就是 Velox 與 Prestissimo:保留 coordinator、connector 與 split 模型,替換掉論文第五節極力辯護的那些 JVM 運算子。Connector API 以及它隱含的 catalog 模型,如今已是查詢引擎接上 Iceberg、Delta Lake 與 Hudi 表格的標準做法,因此現代 lakehouse 那套「開放表格式與可插拔引擎分離」的架構,大體上就是 Presto 架構的一般化。它最持久的想法或許也最樸素:shuffle 是否物化,是一個可與延遲交換的旋鈕,而不是分散式查詢引擎的既定性質。

論文原文 — 逐字引用

“In aggregate, Presto processes hundreds of petabytes of data and quadrillions of rows per day at Facebook.”

§I

“Tests on a sample of production workload from the Batch ETL use case show that lazy loading reduces data fetched by 78%, cells loaded by 22% and total CPU time by 14%.”

§V-D

“However, as of late 2018, Presto does not have any meaningful built-in fault tolerance for coordinator or worker node crash failures.”

§IV-G

術語 — 依本篇論文的用法

Connector API
Presto 連接外部資料儲存的插件契約,由 Metadata API、Data Location API、Data Source API 與 Data Sink API 四部分組成。它的設計目標是讓 connector 實作在實體分散的執行引擎環境中仍能維持高效能。
Split
指向外部儲存系統中某塊可定址資料的不透明控制代碼,內容由 connector 定義——檔案系統是路徑加位移量,Redis 則是鍵、值格式與主機清單。split 既是指派給 leaf stage task 的單位,也是 worker 執行緒上的排程單位。
Stage 與 task
stage 是計畫中可跨 worker 平行執行的部分;每個 stage 會分散成一或多個 task,各自對不同的輸入資料執行相同的運算。leaf stage 從 connector 讀取資料,intermediate stage 只消費其他 stage 的中間結果。
Pipeline 與 driver loop
pipeline 是 task 內部的一串運算子,例如 hash join 的 build pipeline 與 probe pipeline,並透過本機的 in-memory shuffle 與其他 pipeline 相連。driver loop 執行一個 split 的方式,是反覆在所有能推進的運算子配對之間搬動資料,直到 quanta 用盡或無法再推進為止。
Page 與 Block
page 是 driver loop 在運算子之間搬動的資料單位:一段資料列序列的欄式編碼,每欄一個 Block,且採扁平的記憶體表示。Block 可以是一般、dictionary 編碼或 run-length 編碼;扁平之所以重要,是因為指標追逐、unboxing 與虛擬方法呼叫會主宰緊湊迴圈的成本。
Data layout
connector 向最佳化器揭露的資料表實體描述——位置資訊,加上分割、排序、grouping 與索引等屬性。一個資料表可以回傳多個 layout,讓最佳化器挑選最適合該查詢的一個,例如在述詞欄位上建有索引的 layout。
User memory 與 system memory
user memory 是使用者只憑對查詢與輸入資料的基本認識就能推理的用量,例如聚合的記憶體與其基數成正比;system memory 則是實作的副產物,例如 shuffle 緩衝區,可能與查詢形狀無關。Presto 對 user memory 以及 user 加 system 的總量分別設限。
Reserved pool
查詢記憶體在每個節點上再切出的一塊子 pool,用來在 general pool 耗盡且無法 spilling 時解開叢集的僵局。記憶體吃緊節點上用量最大的查詢會被提升到所有 worker 的 reserved pool,而全叢集同時只允許一個查詢佔用,以免不同 worker 各自卡住不同查詢而形成死結。
Raptor
專為 Presto 撰寫的 shared-nothing 儲存引擎,metadata 放在 MySQL,資料以 ORC 格式存於本機快閃磁碟,並支援排序、bucketing 與時間欄位。它支撐 A/B Testing 使用情境,因為該場景重視可預測的高吞吐、低延遲讀取,勝過就地查詢原始資料。

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

在時間軸上查看