Presto系列 | Presto基本介紹

前言

Presto是一款Facebook開源的MPP架構(gòu)的OLAP查詢引擎，可針對(duì)不同數(shù)據(jù)源執(zhí)行大容量數(shù)據(jù)集的一款分布式SQL執(zhí)行引擎。因?yàn)楣ぷ髦薪佑|到Presto，研究它對(duì)理解SQL Parser、常見算子的實(shí)現(xiàn)（如SQL中table scan,join,aggregation）、資源管理與調(diào)度、查詢優(yōu)化（如向量化執(zhí)行、動(dòng)態(tài)代碼生成）、大數(shù)據(jù)下各個(gè)組件為何適用不同場(chǎng)景等等都有幫助。我希望通過這個(gè)系列可以了解一條SQL在大數(shù)據(jù)場(chǎng)景下該如何高效執(zhí)行。233醬準(zhǔn)備不定時(shí)持續(xù)更新這個(gè)系列，本文主要從Presto的使用舉例，Presto的應(yīng)用場(chǎng)景、Presto的基本概念三個(gè)部分來初步介紹Presto。

Presto的使用舉例

比如說，你想對(duì)存儲(chǔ)在不同數(shù)據(jù)源中的數(shù)據(jù)，如HDFS、Mysql、HBase等通過一個(gè)SQL做查詢分析，那么只需要把每一個(gè)數(shù)據(jù)源當(dāng)成是Presto的Connector，對(duì)應(yīng)實(shí)現(xiàn)Presto SPI暴露出的Connector API就可以了。

Presto官方版和Presto社區(qū)版已經(jīng)支持了很多Connector，社區(qū)版略勝一籌。至于兩者有何區(qū)別，吃瓜群眾可以前往文末參考資料[4]。簡(jiǎn)而言之，都主要由Facebook那幫大佬核心維護(hù)。社區(qū)版更新更為頻繁，但高版本需要JDK11才能支持；官方版JDK8就行，官方版的Star數(shù)是社區(qū)版的10倍左右，選哪個(gè)就一目了然了吧。

Presto的應(yīng)用場(chǎng)景

Presto是為了處理TB/PB級(jí)別的數(shù)據(jù)查詢和分析，它是OLAP(Online Analytical Processing)領(lǐng)域的一個(gè)計(jì)算引擎。參考資料[1]提到了Presto在Facebook中的使用場(chǎng)景有：

報(bào)表和大盤查詢

做過報(bào)表和大盤的小伙伴應(yīng)該對(duì)這個(gè)場(chǎng)景下復(fù)雜的SQL有所了解。這個(gè)場(chǎng)景下的使用用戶是Facebook內(nèi)部或外部人員，通常要求：高QPS，低時(shí)延(<1s)。

Adhoc分析

Ad hoc是拉丁文「for this purpose」的意思，Adhoc query的查詢特點(diǎn)是海量、實(shí)時(shí)、靈活。數(shù)據(jù)量如PB級(jí)別以上，時(shí)延秒-分鐘級(jí)別，靈活性舉例子如下：

var?adhoclQuery?=?"SELECT?*?FROM?table?WHERE?id?=?"?+?myId;

var?sqlQuery?=?"SELECT?*?FROM?table?WHERE?id?=?1";

adhoclQuery的結(jié)果取決于參數(shù)“myId”的值，它的結(jié)果不能被預(yù)計(jì)算。sqlQuery的結(jié)果每次執(zhí)行可認(rèn)為都一樣，它的結(jié)果可以被預(yù)計(jì)算。

典型應(yīng)用場(chǎng)景如：用戶趨勢(shì)分析，產(chǎn)品市場(chǎng)洞察等。主要用戶是內(nèi)部數(shù)據(jù)分析人員。

批處理

批處理通常是指更大數(shù)據(jù)量的一個(gè)分析，可容忍高時(shí)延(小時(shí)-天級(jí)別)。Presto是為了低時(shí)延而設(shè)計(jì)的，它屬于內(nèi)存型的MPP架構(gòu)。并不適合類似Spark那樣的長(zhǎng)時(shí)間離線跑批。參考資料[1]的視頻中分析了兩者架構(gòu)的區(qū)別，Presto跑批的限制。這里我截幾張PPT幫助大家理解：
兩者的架構(gòu)區(qū)別：

所以他們提供了Presto on Spark方案，這樣做的好處是可以統(tǒng)一用戶使用的SQL方言差異，UDF差異。

當(dāng)然，業(yè)界除了Facebook還有公司把Presto跑在Spark上來跑批嗎？我沒有搜到相關(guān)信息。

Presto的基本概念

前面主要談了Presto的使用場(chǎng)景，下面簡(jiǎn)要從 Presto的架構(gòu)和基本術(shù)語上介紹Presto。

Presto架構(gòu)

Presto的架構(gòu)圖如下：

Coordinator節(jié)點(diǎn) 負(fù)責(zé)接受客戶端請(qǐng)求、解析SQL語句、生成并優(yōu)化分布式邏輯執(zhí)行計(jì)劃、將計(jì)劃中的任務(wù)調(diào)度到Worker節(jié)點(diǎn)上，并跟蹤Worker節(jié)點(diǎn)和任務(wù)的執(zhí)行狀態(tài)。

Worker節(jié)點(diǎn) 負(fù)責(zé)任務(wù)的執(zhí)行，接受Coordinator節(jié)點(diǎn)的調(diào)度。

從中我們可以粗略看出一條SQL在Presto中的執(zhí)行過程為：

1).Client發(fā)送一個(gè)SQL語句到Coordinator節(jié)點(diǎn)
2).Coordinator節(jié)點(diǎn)把請(qǐng)求放到隊(duì)列中，解析和分析其中的SQL語句；生成并且優(yōu)化分布式邏輯執(zhí)行計(jì)劃。
3).Coordinator節(jié)點(diǎn)會(huì)把這個(gè)Plan分解為任務(wù)，由多個(gè)Worker分布式執(zhí)行。

要想了解具體的SQL執(zhí)行過程，我們得先介紹下Presto的基本概念，也為下篇介紹「Presto為什么是OLAP領(lǐng)域的實(shí)時(shí)計(jì)算引擎」的文章作準(zhǔn)備>_<

基本術(shù)語

我們很容易知道 statements 和 queries 的意思。作為一個(gè)使用者我們也應(yīng)該熟悉 stages、 splits 這些概念使Presto盡可能高效執(zhí)行queries；作為一個(gè)Presto管理員，應(yīng)該理解 stages 是如何映射為tasks的，包含 drivers 集合 的 task 是如何處理數(shù)據(jù)的。以下將從一般到具體介紹Presto的基本術(shù)語。

Server Types

Presto包含兩種類型的服務(wù)端節(jié)點(diǎn)：coordinators 和 workers。

Coordinator

Presto中的Coordinator節(jié)點(diǎn)負(fù)責(zé)解析SQL語句，生成并優(yōu)化物理執(zhí)行計(jì)劃，管理Presto worker節(jié)點(diǎn)。它是Presto運(yùn)行的“大腦”。它也是客戶端提交SQL語句的節(jié)點(diǎn)。每個(gè)運(yùn)行的Presto集群包含1個(gè)Coordinator節(jié)點(diǎn)和1-多個(gè)Worker節(jié)點(diǎn)。一個(gè)服務(wù)示例可同時(shí)擔(dān)任這兩種節(jié)點(diǎn)角色。

Coordinator節(jié)點(diǎn)一直跟蹤每個(gè)Worker節(jié)點(diǎn)的狀態(tài)和協(xié)調(diào)查詢計(jì)劃的執(zhí)行。Coordinator生成一個(gè)物理執(zhí)行計(jì)劃模型，它包含一系列的stages，而stages會(huì)轉(zhuǎn)化為一系列的任務(wù)跑在workers節(jié)點(diǎn)上。

Coordinator通過REST API和workers 、 clients通信。

Worker

Worker節(jié)點(diǎn)負(fù)責(zé)執(zhí)行tasks，處理數(shù)據(jù)。它從connectors中撈取數(shù)據(jù)，并且Worker節(jié)點(diǎn)之間可交換中間數(shù)據(jù)。coordinator節(jié)點(diǎn)負(fù)責(zé)合并workers的結(jié)果，并且返回結(jié)果給Client。

當(dāng)一個(gè)Worker節(jié)點(diǎn)開始工作后，它會(huì)把自己注冊(cè)到coordinator的注冊(cè)服務(wù)上，從而使Coordinator節(jié)點(diǎn)可將task調(diào)度到自己執(zhí)行。

Workers和其他Workers、coordinators之間都是通過REST API通信。

Data Sources

諸如connector, catalog, schema, and table這些術(shù)語，都是和Presto的模型中：一種特定的數(shù)據(jù)源有關(guān)。

Connector

connector是Presto中的一個(gè)數(shù)據(jù)源，可以是Hive、Mysql、Elasticsearch、HBase等。你可以把connector認(rèn)為是一種數(shù)據(jù)庫驅(qū)動(dòng)，只要實(shí)現(xiàn)Presto ?SPI 中暴露的相關(guān)接口，就可以接入一種Connector。

Presto自帶一些connectors：如JMX，System connector用來獲取system tables的，Hive connector，TPCH connector 用來性能測(cè)試用的，等等。

每一個(gè)catalog和一個(gè)特定的connector關(guān)聯(lián)。每一個(gè)catalog配置文件中有一個(gè)connector.name屬性，它是被catalog manager用來為一個(gè)給定的catalog創(chuàng)建一個(gè)connector。一個(gè)catalog可以使用相同的connector獲取類似數(shù)據(jù)庫的兩個(gè)實(shí)例。

Catalog

Presto catalog包含schemas和通過Connector持有的數(shù)據(jù)源引用。比如：你可以配置一個(gè)ES catalog，就可以通過ES Connector提供從ES中獲取數(shù)據(jù)。

#elasticsearch.properties

connector.name=elasticsearch
elasticsearch.host=es?host
elasticsearch.port=9200
elasticsearch.default-schema-name=es

當(dāng)你在Presto上執(zhí)行SQL時(shí)，你就在運(yùn)行1-多個(gè)catalogs.在Presto上定位一張表，是通過一個(gè)catalog的全限定名確定的，如hive.test_data.test代表在hive catalog，test_data schema 下的一張test table.

Catalogs屬性文件是存儲(chǔ)在Presto配置目錄的，默認(rèn)是Presto主安裝文件下的etc目錄下。

Schema

Schemas是一種組織tables的方式。一個(gè)catalog和一個(gè)catalog定義了一個(gè)可被查詢的table集合。對(duì)于MySQL這種關(guān)系型數(shù)據(jù)庫，Presto的schema是和MySQL中的schema相同的概念。對(duì)于其他類型的connector，如ES, Presto的schema是用來組織一些表到特定的schema中，從而使底層的數(shù)據(jù)源能夠在Presto層面說得通。

Table

table是一組無序的Row集合，Row是一組有類型的column集合。和關(guān)系型數(shù)據(jù)庫中的概念一樣，table的映射是由connector中定義的。

Query Execution Model

Presto執(zhí)行SQL語句，并且轉(zhuǎn)化為執(zhí)行計(jì)劃，在由coordinator 和 workers組成的分布式集群上運(yùn)行。

Statement

Presto執(zhí)行兼容ANSI標(biāo)準(zhǔn)的SQL。這些SQL statements包含子句，表達(dá)式，條件。
Presto把Statement 和 Query區(qū)分開是因?yàn)椋涸赑resto中，statements是指Client提交上來的SQL語句，如：

SELECT?*?FROM?table?WHERE?id?=?1

query是指Presto執(zhí)行statement時(shí)，生成的一個(gè)物理執(zhí)行計(jì)劃,并且之后分布式的在一系列workers上執(zhí)行它。

Query

當(dāng)Presto解析一個(gè)statement時(shí)，它會(huì)把statement轉(zhuǎn)化為一個(gè)query，并且創(chuàng)建一個(gè)分布式的執(zhí)行計(jì)劃，然后轉(zhuǎn)化為一系列的有關(guān)聯(lián)性的stages運(yùn)行在Presto workers上。當(dāng)你在Presto獲取一個(gè)query的信息時(shí)，得到的是每個(gè)參與執(zhí)行的組件的一個(gè)當(dāng)前結(jié)果快照。

statement可認(rèn)為是Client提交上來的SQL語句，query指的是執(zhí)行一個(gè)statement有關(guān)的配置和組件實(shí)例信息。query圍繞著stages, tasks, splits, connectors，其他組件和數(shù)據(jù)源一起工作，以產(chǎn)生最終結(jié)果。

Stage

當(dāng)Presto執(zhí)行一個(gè)query時(shí)，它會(huì)把執(zhí)行分為一個(gè)有層次結(jié)構(gòu)關(guān)系的stages.比如SQL語句：

這個(gè)層次結(jié)構(gòu)的stages可以理解為一個(gè)一個(gè)樹。每個(gè)query都有一個(gè)root stage負(fù)責(zé)其他stages的輸出結(jié)果聚合。stages是coordinator節(jié)點(diǎn)用來生成一個(gè)分布式查詢計(jì)劃的模型，但是stages它們自己并不跑在Presto workers節(jié)點(diǎn)上。

Task

一個(gè)stage是由一系列的tasks分布式運(yùn)行在Presto workers上。

在Presto架構(gòu)中，task是“work horse”。因?yàn)榉植际讲樵冇?jì)劃被分解為一系列stage，然后被轉(zhuǎn)換為task，這些task隨后執(zhí)行或被進(jìn)一步split。一個(gè)task有輸入和輸出，就像一個(gè)stage可以有一系列的tasks并行執(zhí)行一樣，一個(gè)task可以由一系列的drivers并行執(zhí)行。

Split

Split是較大數(shù)據(jù)集的一個(gè)分片。分布式查詢計(jì)劃的最低級(jí)別的stage(如上圖中的Stage3/Stage4)通過來自connectors得到的splits集合獲取輸入數(shù)據(jù)，更高級(jí)別的中間Stage(如上圖中的Stage2/Stage1)從下一層stage中獲取輸入數(shù)據(jù)。

當(dāng)Presto調(diào)度一個(gè)query時(shí)，coordinator節(jié)點(diǎn)會(huì)查詢連接器的SPI接口獲得一個(gè)表可用的所有split集合。coordinator跟蹤哪些機(jī)器正在運(yùn)行哪些task，以及哪些任務(wù)正在處理哪些split。

Driver

Task包含一個(gè)或多個(gè)并行的driver。driver對(duì)數(shù)據(jù)進(jìn)行操作，并結(jié)合operators產(chǎn)生輸出，然后結(jié)果由一個(gè)task聚合，然后傳遞到另一個(gè)stage的另一個(gè)task。driver是一系列operators實(shí)例，或者您可以將driver看作內(nèi)存中的operator的物理集合。它是Presto體系結(jié)構(gòu)中并行的最低級(jí)別。一個(gè)driver有一個(gè)輸入和一個(gè)輸出。

Operator

一個(gè)operator消費(fèi)、轉(zhuǎn)換和生產(chǎn)數(shù)據(jù)。例如，一個(gè)table scan operator從一個(gè)connector中獲取數(shù)據(jù)并生產(chǎn)出可由其他operator消費(fèi)的數(shù)據(jù)，一個(gè)filter operator通過對(duì)輸入數(shù)據(jù)應(yīng)用謂詞(過濾條件)并生成一個(gè)子集。

Exchange

Exchanges為一個(gè)query的不同stage在Presto節(jié)點(diǎn)之間傳遞數(shù)據(jù)。task使用一個(gè)exchange client，生產(chǎn)數(shù)據(jù)到一個(gè)output buffer中，并且消費(fèi)其他task產(chǎn)生的數(shù)據(jù)。

參考資料

[1].https://databricks.com/session_na20/presto-on-apache-spark-a-tale-of-two-computation-engines

[2].https://research.fb.com/wp-content/uploads/2019/03/Presto-SQL-on-Everything.pdf

[3].https://prestodb.io/docs/current/overview/concepts.html