概要速覽
PrestoDB的Aria項目曾於2020年發佈過一組實驗性功能，用來提高對錶（通過Hive連接器連接並以ORC格式存儲數據）的掃描性能。
在本文中，我們將在基於Docker的PrestoDB測試環境中對這些新功能進行基礎性的測試。[1]
Presto
Presto 是一款能夠大規模並行處理 (MPP) 的SQL執行引擎。執行引擎與數據存儲是分離的，該項目包含大量插件（又稱為連接器，connector），它們為Presto引擎提供查詢的數據。數據存儲中的數據被讀取後，交由Presto執行查詢操作，比如數據連接(joining)和聚合(aggregation)。這種數據存儲和執行分離的架構允許單個Presto實例查詢多個數據源，從而提供了非常強大的聯合查詢層。
Presto有許多可用的連接器，社區也會定期提供用以訪問數據存儲的新型連接器。
Hive 連接器
Hive 連接器一般被視為Presto的標準連接器。我們通常用它連接到 Hive Metastore，以此來獲取Metastore 中定義的表的元數據信息。數據通常存儲在 HDFS 或 S3中，而Metastore提供有關文件存儲位置和格式的信息；最常用的是ORC格式，但也支持 Avro 和 Parquet等其他格式。Hive 連接器允許 Presto 引擎並行地將數據從HDFS/S3掃描到引擎中來執行查詢。ORC格式是一種非常標準且常見的數據存儲格式，能提供很好的壓縮比和性能。
兩個用於執行查詢的核心服務
Presto有兩個用於執行查詢的核心服務：一個負責查詢解析和任務調度等職責的Coordinator，以及多個負責並行執行查詢的Worker。理論上，Coordinator也可以充當Worker的角色，但在生產環境中不會這麼操作。鑑於我們在這裏測試的是Presto，為方便起見，我們只使用一個節點，既作為 Coordinator 也作為 Worker。[2]
我們將使用單個Docker容器來進行本次Presto的測試。請點擊查看部署文檔，文檔的末尾處有如何實現單節點 Presto部署的示例。
下面來介紹Presto是如何執行一條查詢語句的：
首先，Presto coordinator先對查詢語句進行解析，從而制定出一個執行計劃（下文會提供示例展示）。計劃制定完成之後就會被分成幾個階段（或片段），每個階段將執行一系列操作，即引擎用來執行查詢的特定函數。執行計劃通常從連接器掃描數據開始，然後執行一系列操作，如數據過濾、部分聚合以及在Presto worker節點之間交換數據來執行數據連接和最終的數據聚合等。所有這些階段被分成多個分片（split)，即Presto中的並行執行單元。Worker 並行執行可配置數量的分片，從而獲得所需的結果。引擎中的所有數據都保存在內存中（前提是不超過集羣的容量閾值）。
Hive連接器（以及所有其他連接器）負責將輸入數據集拆分為多個分片，供 Presto 並行讀取。作為一項優化措施，Presto 引擎將告知連接器查詢中使用的謂詞（predicate)以及選定的列（column)——稱為謂詞下推 (predicate pushdown)，這使得連接器能夠在把數據提供給Presto引擎之前過濾掉不必要的數據，這也是本文的重點所在。
為了演示謂詞下推，我們來看一個基本查詢——統計某個數據表內符合條件的行數。我們的查詢示例是基於基準測試數據集TPC-H的lineitem數據表進行的。TPC-H的lineitem表中大約有6億行記錄，它們的shipdate字段取值介於1992和1998之間。下面的查詢語句是針對lineitem數據表的設置條件過濾謂詞，篩選出shipdate字段為1992年的數據行。我們先在不啓用Aria增強會話屬性的情況下，通過運行 EXPLAIN 命令來觀察一下查詢計劃：
presto:tpch> EXPLAIN (TYPE DISTRIBUTED) SELECT COUNT(shipdate) FROM lineitem WHERE shipdate BETWEEN DATE '1992-01-01' AND DATE '1992-12-31';
Fragment 0 [SINGLE]
Output layout: [count]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
- Output[_col0] => [count:bigint]
_col0 := count
- Aggregate(FINAL) => [count:bigint]
count := ""presto.default.count""((count_4))
- LocalExchange[SINGLE] () => [count_4:bigint]
- RemoteSource[1] => [count_4:bigint]
Fragment 1 [SOURCE]
Output layout: [count_4]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
- Aggregate(PARTIAL) => [count_4:bigint]
count_4 := ""presto.default.count""((shipdate))
-ScanFilter[table = TableHandle {connectorId='hive', connectorHandle='HiveTableHandle{schemaName=tpch, tableName=lineitem, analyzePartitionValues=Optional.empty}', layout='Optional[tpch.lineitem{domains={shipdate=[ [[1992-01-01, 1992-12-31]] ]}}]'}, grouped = false, filterPredicate = shipdate BETWEEN (DATE 1992-01-01) AND (DATE 1992-12-31)] => [shipdate:date]
Estimates: {rows: 600037902 (2.79GB), cpu: 3000189510.00, memory: 0.00, network: 0.00}/{rows: ? (?), cpu: 6000379020.00, memory: 0.00, network: 0.00}
LAYOUT: tpch.lineitem{domains={shipdate=[ [[1992-01-01, 1992-12-31]] ]}}
shipdate := shipdate:date:10:REGULAR查詢計劃按自下而上順序來閲讀，從 Fragment 1 開始，並行掃描 lineitem 表，使用謂詞對shipdate列進行過濾，然後對每個分片執行部分聚合，並將該部分結果交換到下一階段 Fragment 0 來執行最終的聚合，之後再將結果發送到客户端，查詢計劃流程參見下圖：（圖中靠近底部的水平線標示出哪些代碼在 Hive 連接器中執行，哪些代碼在 Presto 引擎中執行。）
現在我們來執行這個查詢！
presto:tpch> SELECT COUNT(shipdate) FROM lineitem WHERE shipdate BETWEEN DATE '1992-01-01' AND DATE '1992-12-31';
_col0
----------
76036301
(1 row)
Query 20200609_154258_00019_ug2v4, FINISHED, 1 node
Splits: 367 total, 367 done (100.00%)
0:09 [600M rows, 928MB] [63.2M rows/s, 97.7MB/s]我們看到，lineitem 表包含7600多萬行shipdate列取值為1992年的記錄。執行這個查詢大約花費了9 秒，總共處理了 6 億行數據。
現在我們來激活會話屬性 pushdown_subfields_enabled 和hive.pushdown_filter_enabled，以啓用 Aria 功能，下面我們來看一下查詢計劃發生了怎樣的變化：
presto:tpch> SET SESSION pushdown_subfields_enabled=true;
SET SESSION
presto:tpch> SET SESSION hive.pushdown_filter_enabled=true;
SET SESSION
presto:tpch> EXPLAIN (TYPE DISTRIBUTED) SELECT COUNT(shipdate) FROM lineitem WHERE shipdate BETWEEN DATE '1992-01-01' AND DATE '1992-12-31';
Fragment 0 [SINGLE]
Output layout: [count]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
- Output[_col0] => [count:bigint]
_col0 := count
- Aggregate(FINAL) => [count:bigint]
count := ""presto.default.count""((count_4))
- LocalExchange[SINGLE] () => [count_4:bigint]
- RemoteSource[1] => [count_4:bigint]
Fragment 1 [SOURCE]
Output layout: [count_4]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
- Aggregate(PARTIAL) => [count_4:bigint]
count_4 := ""presto.default.count""((shipdate))
- TableScan[TableHandle {connectorId='hive', connectorHandle='HiveTableHandle{schemaName=tpch, tableName=lineitem, analyzePartitionValues=Optional.empty}', layout='Optional[tpch.lineitem{domains={shipdate=[ [[1992-01-01, 1992-12-31]] ]}}]'}, grouped = false] => [shipdate:date]
Estimates: {rows: 540034112 (2.51GB), cpu: 2700170559.00, memory: 0.00, network: 0.00}
LAYOUT: tpch.lineitem{domains={shipdate=[ [[1992-01-01, 1992-12-31]] ]}}
shipdate := shipdate:date:10:REGULAR
:: [[1992-01-01, 1992-12-31]]注意：查詢計劃的主要變化位於底部，即TableScan 操作中包含了shipdate 列。連接器已經接收到shipdate列上的謂詞條件——取值介於1992-01-01 和 1992-12-31之間。如下圖所示，該謂詞被下推到連接器，免去了查詢引擎過濾這些數據的必要性。
我們再一次運行這個查詢！
presto:tpch> SELECT COUNT(shipdate) FROM lineitem WHERE shipdate BETWEEN DATE '1992-01-01' AND DATE '1992-12-31';
_col0
----------
76036301
(1 row)
Query 20200609_154413_00023_ug2v4, FINISHED, 1 node
Splits: 367 total, 367 done (100.00%)
0:05 [76M rows, 928MB] [15.5M rows/s, 189MB/s]運行查詢後，我們得到了相同的結果，但查詢時間幾乎縮短了一半，更重要的是，查詢只掃描了7600萬行！連接器已經將謂詞應用於shipdate 列，而不是讓引擎來處理謂詞，因此節省了CPU週期，繼而加快了查詢速度。針對不同的查詢和數據集情況可能有所不同，但如果是通過Hive連接器查詢ORC文件的場景，該方案絕對值得一試。
文章作者：Adam Shook
原文於2020年6月15日發表在作者的個人博客上：http://datacatessen.com
參考
- ^如需瞭解有關Aria項目功能的更多信息，可查看文章底部 https://engineering.fb.com/20...
- ^[2]如需瞭解安裝等詳細信息，可查看文章底部文檔 https://prestodb.io/docs/curr...
