歡迎光臨
每天分享高質量文章

解讀來自Google的程式必備最新高科技-從 F1 Query 論文看 SQL 查詢的執行樣式

導語:F1是Google的分佈式資料庫,問世以來一直受到大家的關註。其中分佈式查詢引擎怎麼實現,也一直是資料庫界最關心的問題之一。F1團隊在VLDB2018上發表了論文詳細論述該話題。本文是對該問題的詳細剖析,十分值得架構師和資料庫從業人員學習。

F1 是起源於 Google AdWords 的分佈式 SQL 查詢引擎,跟底下的 Spanner 分佈式儲存搭配,開啟了分佈式關係資料庫——所謂 NewSQL 的時代。我們今天說的是 F1 團隊在 VLDB2018 上發的文章 F1 Query: Declarative Querying at Scale [1],它和之前我們說的 F1 幾乎是兩個東西。

F1 Query 是一個分佈式的 SQL 執行引擎,現在大資料領域流行的 Presto、Spark SQL、Hive 等等,都可以算在這個範疇里。類似地,F1 Query 也支持對各種不同資料源的查詢,既可以是傳統的關係表、也可以是 Parquet 這樣的半結構化資料。

這樣一來,不同資料格式的壁壘也被打破了:你可以在一個系統里完成對不同資料源的 Join,無論資料以什麼形式存放在哪裡。商業上管這個叫 Federated Query 或者 DataLake,幾家雲計算巨頭都有類似的產品。

那 F1 Query 的貢獻在哪裡呢?

F1 Query 定義了三種不同型別的查詢執行樣式,根據查詢的資料量大小或執行時間,將用戶查詢劃分成:

  1. 單機執行(Centralized Execution)

  2. 分佈式執行(Distributed Execution)

  3. 批處理執行(Batch Execution)

前兩個是交互式的,即客戶端會等待結果傳回。最後一個批處理更像是 ETL:客戶端輸入任務之後就不再管了,查詢結果會被寫到指定的地方。

單機執行

單機執行對應我們熟悉的 OLTP 查詢,例如單表點查、帶索引的 Join 等。這類查詢本身已經足夠簡單,只需幾毫秒就能做完,處理它們的最好方式就是在收到請求的機器上立即執行。

在 F1 Query 系統中有 F1 Server 和 F1 Worker 等角色。F1 Server 負責接收客戶端請求,如果它判斷這個查詢應當使用單機而不是分佈式方式執行,它就親力親為、直接執行並傳回結果。

這樣的行為和絕大多數單機 OLTP 資料庫是一致的,例如 MySQL 採用的是 Thread Pool + Dispatcher 的處理模型,Thread Pool 的規模是一定的,Dispatcher 根據高低優先級分派執行任務。最終一個請求只會被一個執行緒處理,換句話說,對某個查詢來說其執行過程是單執行緒的。

▲ MySQL 的執行緒池處理模型,一般存在多個 Thread Group,圖中描繪了一個 Thread Group

F1 Query 單機查詢的執行器同樣也是教科書式的 Valcano 模型,但也無可厚非——對 OLTP 來說這已經足夠好。如下圖所示,從頂層算子開始遞迴地呼叫 GetNext(),每次取出一行資料,直到沒有更多資料為止。各個算子只需要實現 GetNext() 接口即可,簡單清晰。

分佈式執行

F1 Query 對更複雜的查詢,例如沒有索引的 Join 或聚合等,則採取分佈式查詢的方式。大部分 OLAP 查詢、尤其是 Ad-hoc 的查詢都落在這一分類中。這種情況下,分佈式導致的網絡、調度等 Overhead 已經遠小於查詢本身的成本;而且隨著資料量的增加,單節點記憶體顯然不夠用了。

▲ *F1 Query 的系統架構,主要包含 F1 Master、F1 Server、F1 Worker 三個角色,其他 Catalog、UDF Server、Batch Metadata 用於儲存查詢相關的 Metadata 等* 

這時,上圖中的 F1 Worker 就派上用場了。F1 Server 此時僅僅作為協調者存在,將任務分配給多個 Worker,直到 Worker 的任務全都完成,再把結果彙總發給客戶端。

這個樣式眼熟嗎?你可能會想到 Greenplum 這類的資料倉庫,已經很接近了。最相似的我認為是 Presto。Presto 是 Facebook 開發的一套分佈式 SQL 引擎,如果單單隻看 F1 Query 的分佈式查詢,和 Presto 大同小異。

與單機執行不同的是,分佈式查詢中的算子可以有多個實體(Instance)並行執行,每個實體負責其中一部分資料。在 F1 Query 里這樣一個資料分片被稱為 Fragment,在 Spark SQL 里叫 Partition,在 Presto 里叫 Split。

下麵的例子是一個 Join-Aggregation-Sort 的查詢,它分成了 4 個階段:

  1. Scan(Clicks) 被分配給 1000 個 F1 Worker 上並行拉取資料,並根據每一行資料的 Hash(AdID) 發送給對應的 HashJoin 分片,即一般說的 shuffle 過程;

  2. Scan(Ads) 被分配給 200 個 F1 Worker 上並行拉取資料,並且也以同樣的方式做 shuffle;

  3. HashJoin 及 PartialAggregation:根據 Join Key 分成了 1000 個並行任務,各自做 Join 計算,並做一次聚合;

  4. 最後,F1 Server 把各個分片的聚合結果再彙總起來,傳回給客戶端。

Presto 具有的缺陷,F1 Query 分佈式查詢同樣也有,比如:

  • 純記憶體的計算方式,無法利用磁盤的儲存空間,某些查詢可能面臨記憶體不足;

  • 沒有 Fault-tolerance,對於一個涉及上千台 Worker 的查詢,任何一臺的重啟都會導致查詢失敗。

批處理執行

F1 Query 還有個獨特的批處理執行,這個樣式定位於更大的資料量、更久的查詢時間;另一方面,它的結果不再是傳回給客戶端,而是將查詢結果寫到指定的地方,例如 Colossus(第二代 GFS)上。

上一節我們提道,Presto 的樣式沒有 Fault-tolerance,這對於長時間運行的批處理任務是致命的,查詢失敗的概率會大大增加。批處理查詢首先要解決的就是 Fault-tolerance 問題:必須能以某種方式從 Worker 節點的失敗中恢復

解決這個問題有兩條路可走:一是 MapReduce 的樣式,將計算分成若干個階段(Stage),而中間結果持久化到 HDFS 這樣的分佈式檔案系統上;二是 Spark RDD 樣式,通過記錄祖先(Lineage)信息,萬一發生節點失敗,就通過簡單的重算來恢復丟失的資料分片,這樣資料就可以放在記憶體里不用落盤。

Spark 的做法顯然是更先進的,原因有很多,這裡只說最重要的 2 條。欲知詳情可以看我之前的博客文章《一文讀懂 Apache Spark》[2]。

  1. Spark 的計算基本在記憶體中,只有當記憶體不夠時才會上限溢位到磁盤,而 MR 的中間結果必須寫入外部檔案系統;

  2. Spark 可以把執行計劃 DAG 中相互不依賴的 Stage 並行執行,而 MR 只能線性地一個接一個 Stage 執行。

但是出乎意料的是,F1 Query 採用的是前者,也就是 MR 樣式。這其中的原因我們不得而知,我猜想和 Google 自家的 FlumeJava 不夠給力有關係。

如下圖。左邊的 Physical Plan 和上一節的分佈式查詢是一樣的,不同之處是在批處理樣式下,它被轉換成一系列的 MR 任務,之後交給調度器(Scheduler)去處理即可。

相比分佈式執行,批處理樣式下各個步驟都會持久化到外部檔案系統(因為 MapReduce 的特性所致)。不僅如此,Pipeline 式的執行也沒法進行。以上一節提到的 HashJoin 為例,左邊 Clicks 的 Scan 和 HashJoin 原本是可以 Pipeline 執行的,但是在批處理樣式下,必須等到 Scan(Clicks) 這個階段完成才能進行下一步的 HashJoin 階段。

單機並行執行

除了上面聊的 F1 Query 所支持的 3 種查詢樣式之外,事實上還有一種處理模型位於單執行緒執行和分佈式執行之間:單機的並行執行。初看這似乎與分佈式執行很相似,但又有些不同:

  • 不用考慮單個 Worker 的失敗恢復,因為它們都在同一個行程里;

  • 各個 Worker 執行緒的記憶體是共享的,它們之間交換資料無需考慮網絡通訊代價。

這種樣式在傳統的關係型資料庫上很常見,尤其是 Postgres、SQL Server 這類以 OLAP 查詢見長的選手。以 Postgres 為例,在開啟並行查詢的情況下,查詢優化器會根據代價選擇是否生成並行執行計劃;如果生成了並行執行計劃,執行器會調度多個 Worker 一起完成工作。

下圖是一個 Postgres 上並行 Hash Join 的例子,從執行計划上看和上一節幾乎一樣,但是這裡不再需要對資料做 Shuffle:Hash Join 所用的 Hash Table 本身是全域性共享的。

Parallel Hash Join 並非只有這一種做法。SQL Server 就更接近分佈式執行的方案:把 Hash Key 相同的資料 shuffle 到同一個分片上——但這個 shuffle 只是邏輯上的,不需要真的做 IO。

相比分佈式查詢,單機並行的最大優勢在於響應速度更快,因為省去了大量的網絡 IO 時間,而且調度一個 Worker 執行緒要比調度一個 Worker 機器快得多。

但別忘了,單機運算能力的 scale up 成本非常高,並且是存在上限的。對於 Google 之類的互聯網公司,絕大部分查詢都超出了單機的儲存或計算能力,我猜測這也是 F1 Query 並未考慮單機並行的理由。

對 F1 Query 的評價

從論文描述的情況來看,F1 Query 還不算個完善、成熟的系統,其定位更像是一個解決業務需求的膠水系統,而非 Spanner 這樣的“硬核”技術。它追求的是夠用就好。很多地方其實還有不小的改進空間,舉幾個例子:

  • 對交互式查詢,選擇分佈式還是單機計算目前還是基於啟髮式規則。

  • 三種樣式的執行計劃是用一樣的優化器生成的。但是客觀的說,這其中的差別可是不小的。

  • 優化器是基於規則的。之所以不做 CBO,論文給出的解釋是資料源眾多,不容易估算代價。

  • 批處理樣式下用 Spark 取代 MR 的樣式是更好的選擇。

F1 Query 希望用一套系統解決所有 OLTP、OLAP、ETL 需求、用一套系統訪問資料中心裡各種格式的資料,這兩點才是 F1 Query 的核心競爭力。

SQL 執行樣式總結

從資料庫的視角看,理想的資料庫應當隱藏掉查詢執行的種種細節,只要用戶輸入一個宣告(例如 SQL),就能以最優的方式執行查詢給出答案。F1 Query 做了個勇敢的嘗試,它將多種執行模型揉合在一個系統中,共享同一套優化器和算子,以較低的開發成本獲得其中最優的執行性能(在理想情況下)。

下麵的表格總結了 4 種執行樣式的優勢和不足。

總而言之,所謂 No Free Lunch[3] —— 沒有最優的方案,資料量是決定能選用哪個執行樣式的前提。實踐中,先確保資料量能夠承載的下,再談優化也就明白多了。

References

  1. F1 Query: Declarative Querying at Scale

  2. MySQL Thread Pool Implementation

  3. Presto 實現原理和美團的使用實踐 – 美團技術團隊

文中鏈接:

[1] http://www.vldb.org/pvldb/vol11/p1835-samwel.pdf

[2] https://ericfu.me/apache-spark-in-nutshell/

[3] https://en.wikipedia.org/wiki/No_free_lunch_theorem


關註DRDS樂園公眾號,獲取更多分佈式資料庫相關信息。


活動預告:

11 月 23 ~ 24 日,GIAC 全球互聯網架構大會將於上海舉行。GIAC 是高可用架構技術社區推出的面向架構師、技術負責人及高端技術從業人員的技術架構大會。今年的 GIAC 已經有微軟,騰訊、阿裡巴巴、螞蟻金服,華為,科大訊飛、新浪微博、京東、七牛、美團點評、餓了麽,才雲,格靈深瞳,Databricks,等公司專家出席。


本期 GIAC 大會上,資料庫和大資料部分精彩的議題如下:

參加 GIAC,盤點2018最新技術。點擊“閱讀原文”瞭解大會更多詳情。

赞(0)

分享創造快樂