我有一壶酒，足以慰平生。

0%

presto学习笔记

发表于 更新于 分类于 ， 阅读次数： Disqus：

Presto概述及应用场景

Presto是一种分布式SQL查询引擎，旨在查询分布在一个或多个异构数据源上的大型数据集。例如，查询TB或者PB级别的数据，是查询 HDFS、hive的一个可选项，但同时不局限于 HDFS, presto 适用于 OLAP (On-Line Analytical Processing)联机分析处理) 的场景。presto 并不能用来替代 mysql/pg/oracle，presto 设计的目的也不是用来处理 (OLTP (On-Line Transaction Processing) 联机事务处理) 场景。

Presto架构

架构图

1

节点角色

  • coordinator

    Presto coordinator是负责解析语句，计划查询和管理Presto worker程序节点的服务器。它是Presto安装的“大脑”，也是客户端连接到以提交执行语句的节点。每个Presto安装必须在一个或多个Presto工作人员旁边配备一个Presto coordinator。出于开发或测试目的，可以将Presto的单个实例配置为执行这两个角色。

    coordinator跟踪每个工作人员上的活动并协调查询的执行。coordinator创建涉及一系列阶段的查询的逻辑模型，然后将其转换为在Presto worker群集上运行的一系列关联任务。

    coordinator使用REST API与workers和客户进行通信。

  • worker

    Presto worker是Presto安装中的服务器，负责执行任务和处理数据。辅助节点从连接器获取数据并相互交换中间数据。coordinator负责从worker那里获取结果并将最终结果返回给客户。

    当Presto worker进程启动时，它将自己通告给coordinator中的发现服务器，这使Presto coordinator可以使用它来执行任务。

    worker使用REST API与其他workers和Presto coordinators进行通信。

  • Discovery Service

    presto 使用服务发现来查找集群中所有的节点，每个注册到 Discovery Service 的节点周期性发送心跳信号，这可以使 coordinator 获取最新的可用 worker 节点列表，worker 发送心跳失败，Discovery Service 会触发失败检测，worker 将不会被分配任务。Discovery Service 是内置在 coordinator 节点中。

基于Connector 的架构

只要已有数据在使用 presto 的数据类型时，也可以以表、行、列的形式表达，connector 就可以创建，查询引擎也可以使用这些数据进行查询处理。

Presto 提供了 SPI 来实现一个 connector，通过在 connector 中实现 SPI，Presto 可以使用内部标准操作在任何数据上连接数据源和实施操作。每个 connector 都需要实现这个 API的三个部分：

  • Operations to fetch table/view/schema metadata
  • Operations to produce logical units of data partitioning, so that Presto can parallelize reads and writes
  • Data sources and sinks that convert the source data to/from the in-memory format expected by the query engine

img

查询执行模型(Query Execution Model)

当一个 SQL 语句提交给 coordinator 节点,coordinator 会接收sql 文本，解析并分析它。然后使用被称作查询计划的内部数据结构创建一个用于执行的计划。这个查询计划大体上代表了每个 SQL 语句处理数据并且返回结果所需要的步骤。

img

查询计划生成器使用了 metadata SPI 与 data statistics SPI 来创建查询计划。所以 coordinator 通过直接连接数据源，使用 SPI 来收集关于数据表和其他元数据。

img

coordinator 使用 metadata SPI 来获取表、列和字段类型信息，它们用于验证查询在语义上是否有效，并执行原始查询中表达式的类型检查和安全检查。

statistics SPI 被用来获取关于行数计数和表大小信息，从而在执行计划时实施基于成本的查询优化。

data location SPI 简化了分布式查询计划的创建。它被用来生成表内容的逻辑拆分，split 块是工作分配和并行工作的最小单位。

分布式查询计划是由一个或多个 stage 组成的简单查询计划的扩展。简单的查询计划被分割成计划片段。stage 是计划片段的运行时化身，它包含由 stage 的计划片段描述的工作的所有任务。

coordinator 拆分整个计划，使得集群中的 worker 节点能够并行处理，加速了整个查询过程。拥有多个阶段会导致创建阶段的依赖树。stage 的数量依赖于查询的复杂度。

v2-243a2a1abaa1f76dc17ec7cf248cccd5_720w

分布式查询计划定义了查询在 presto 集群上执行的 stage 和方式。coordinator 用来进一步在 worker 之间计划和调度任务。一个 stage 由一个或多个 任务构成。

img

任务处理的数据单位被称为 split，split 是底层数据段的描述符，可以由工作进程检索和处理。它是并行和工作分配的单位。连接器对数据执行的特定操作取决于基础数据源。例如，hive connector 用文件的路径、offset、程度来描述 split，来表示文件的哪部分是需要被处理的。

任务在源阶段以 pages 的形式处理数据，这是一组列格式的行。这些 pages 流向其他中间下游阶段。Pages 通过交换操作在 stages 之间转换，这是从上游 stage 的任务读取数据。

source 任务在 connector 的帮助下，使用 data source SPI 从底层数据源获取数据，这些数据以页面的形式呈现给Presto并流经引擎。Operators 根据他们的语法，处理并产生 pages。一个 task 操作的序列被称作 pipeline.

pipeline 的最后一个操作通常是放置它的 output pages 到任务的 output buffer。下游任务的交换操作消费来自上游任务 output buffer 的 pages。所有的这个操作并行发生在不同的 worker 中。

img

所以任务被指定到 worker 节点时，是运行时计划片段的化身。当任务被创建后，它给每一个 split 实例化一个 driver。每个 driver 是对 split 中数据操作处理的 pipeline 的实例化。一个 task 可能使用一个或多个 driver，这取决于 presto 的配置。一旦 所有的 driver 结束，数据将会传递到下一个 split，driver 和 task 将随着他们工作的结束被销毁。

img

为了处理查询，协调器将使用连接器中的元数据创建 split 列表。通过使用 split 列表，协调器开始调度 worker 节点上的任务，以收集 split 中的数据。在查询执行期间，协调器跟踪可用于处理的所有 split，以及在工作线程和处理 split 上运行任务的位置。当任务完成处理并为下游处理生成更多split时，协调器将继续调度任务，直到没有 split 可供处理为止。在worker上处理完所有 split后，所有数据都可用，协调器可以将结果提供给 client。

Presto部署安装

下载（0.243.2）

官网

安装

1
tar -zxcf presto-server-0.243.1 -C:/usr/local

配置

在安装目录中创建etc目录。这将具有以下配置：

  • Node Properties: 特定于每个节点的环境配置
  • JVM Config: Java虚拟机的命令行选项
  • Config Properties: Presto服务器的配置
  • Catalog Properties: 连接器的配置（数据源）

目录树如下：

image-20201116143331173

具体配置如下：

node.properties

1
2
3
node.environment = production
node.id = ffffffff-ffff-ffff-ffff-ffffffffffff
node.data-dir = /usr/local/presto-server-0.243.1/data
  • node.environment：环境名称。群集中的所有Presto节点必须具有相同的环境名称。
  • node.id：此Presto安装的唯一标识符。这对于每个节点都必须是唯一的。在重新启动或升级Presto时，此标识符应保持一致。如果在一台计算机上运行多个Presto安装（即同一台计算机上的多个节点），则每个安装必须具有唯一的标识符。
  • node.data-dir：数据目录的位置（文件系统路径）。Presto将在此处存储日志和其他数据。

jvm.config

1
2
3
4
5
6
7
8
-server
-Xmx1G
-XX:+UseG1GC
-XX:G1HeapRegionSize=32M
-XX:+UseGCOverheadLimit
-XX:+ExplicitGCInvokesConcurrent
-XX:+HeapDumpOnOutOfMemoryError
-XX:+ExitOnOutOfMemoryError

设置java虚拟机参数

config.properties(单机版)

1
2
3
4
5
6
7
8
coordinator=true
node-scheduler.include-coordinator=true
http-server.http.port=8848
query.max-memory=1GB
query.max-memory-per-node=128MB
query.max-total-memory-per-node=256MB
discovery-server.enabled=true
discovery.uri=http://192.168.88.110:8848

config.properties(集群版）

  • coordinator

    config.properties(主)

    1
    2
    3
    4
    5
    6
    7
    8
    coordinator=true
    node-scheduler.include-coordinator=false
    http-server.http.port=8080
    query.max-memory=50GB
    query.max-memory-per-node=1GB
    query.max-total-memory-per-node=2GB
    discovery-server.enabled=true
    discovery.uri=http://example.net:8080

  • worker

    config.properties（从）

    1
    2
    3
    4
    5
    6
    coordinator=false
    http-server.http.port=8080
    query.max-memory=50GB
    query.max-memory-per-node=1GB
    query.max-total-memory-per-node=2GB
    discovery.uri=http://example.net:8080
    • coordinator：允许此Presto实例充当协调器（接受来自客户端的查询并管理查询执行）。
    • node-scheduler.include-coordinator：允许在协调器上安排工作。对于较大的群集，协调器上的处理工作会影响查询性能，因为计算机的资源不可用于计划，管理和监视查询执行的关键任务。
    • http-server.http.port：指定HTTP服务器的端口。Presto使用HTTP进行内部和外部所有通信。
    • query.max-memory：查询可以使用的最大分布式内存量。
    • query.max-memory-per-node：查询可在任何一台计算机上使用的最大用户内存量。
    • query.max-total-memory-per-node：查询可在任何一台计算机上使用的最大用户和系统内存量，其中系统内存是读取器，写入器和网络缓冲区等在执行期间使用的内存。
    • discovery-server.enabled：Presto使用发现服务来查找集群中的所有节点。每个Presto实例在启动时都会向Discovery服务注册。为了简化部署并避免运行其他服务，Presto协调器可以运行发现服务的嵌入式版本。它与Presto共享HTTP服务器，因此使用相同的端口。
    • discovery.uri：发现服务器的URI。因为我们已经在Presto协调器中启用了Discovery的嵌入式版本，所以它应该是Presto协调器的URI。更换example.net:8080以匹配Presto协调器的主机和端口。此URI不得以斜杠结尾。
    • 扩展：
      • jmx.rmiregistry.port：指定JMX RMI注册表的端口。JMX客户端应连接到该端口。
      • jmx.rmiserver.port：指定JMX RMI服务器的端口。Presto导出许多可用于通过JMX监视的指标。

log.properties

1
com.facebook.presto=INFO

catalog(配置数据源)

  • jmx.properties

    1
    connector.name = jmx

    ​ Presto通过安装在目录中的连接器访问数据。连接器提供目录内部的所有模式和表。

  • mysql.properties

    1
    2
    3
    4
    connector.name=mysql
    connection-url=jdbc:mysql://192.168.88.110:3306
    connection-user=root
    connection-password=123456

  • hive.properties

    1
    2
    3
    4
    connector.name=hive-hadoop2
    #注意 connector.name 只能是 hive-hadoop2 
    hive.metastore.uri=thrift://192.168.88.110:9083
    hive.config.resources=/etc/hadoop/core-site.xml,/etc/hadoop/hdfs-site.xml

启动

1
2
3
cd /usr/local/presto-server-0.243.1/bin

./launcher start

image-20201116145417445

Presto Web UI

image-20201116145446100

首页

Presto Web UI 可以用来检查和监控Presto集群，以及运行的查询。他所提供的关于查询的详细信息可以更好的理解以及调整整个集群和单个查询。

需要注意的是，Presto Web UI所展示的信息都来自于Presto系统表。

当你进入Presto Web时，你将会看到如同1所示的界面：主要分为上下两部分，上面描述了集群信息，下面是查询列表；

集群信息

Running Queries

当前在集群中正在执行的查询的个数。包含所有用户提交的查询；例如，如果Alice正在执行两个查询，Bob正在执行五个查询，那么在这个指标下显示的是7。

Queued Queries

当前集群队列中正在等待的查询的个数，也是包含所有用户的查询。队列中的查询表示这些查询正在等待Coordinator根据Resource Group的配置为他们安排调度；

Blocked Queries

集群中被阻塞的查询的个数；被阻塞的查询意味着该查询因为缺少可用的Splits或者资源而无法继续执行。

Active Workers

集群中当前活跃的节点的个数；任何手动会自动添加或删除的节点都会注册到Discovery 服务，同时这里展示的数字将会更新。

Runnable Drivers

集群中可运行的Drivers的平均数量（当Task被创建之后，他为每一个Split实例化一个Driver，每一个Driver就是一个Pipeline 中Operators的实例，并对来自Split的数据进行处理，一旦Driver完成，数据将会被传给下一个Split）。

Reserved Memory

集群中Reserved Memory的大小，单位是bytes。

Rows/Sec

集群中所有查询在每一秒钟处理的行数。

Bytes/Sec

集群中所有查询在一秒钟处理的总共的Bytes。

Worker Parallelism

Worker的并发总数，在集群中运行的所有Worker和所有查询的CPU Time总和。

查询列表

WBE UI首页下部分就是查询列表的展示，当前列表中可以展示的查询的数量时可以配置的。如下图所示

image-20201117135230278

你可以根据一些条件过滤和定位你想要的查询；同时提供了搜索输入框用于定位查询，输入的值会匹配很多项，包括：用户名、查询发起人，查询source，查询ID，resource group甚至SQL文本，和查询状态。同样你可以根据后面预设的一些状态（running, queued, finished, and failed）对查询进行筛选；

最左边的控件允许你确定显示的查询的排序顺序、重新排序的时间以及要显示的查询的最大数量。

下面的每一行表示一个查询，左侧如下图所示，右侧为查询的SQL文本；

image-20201117135433029

根据上图可以观察当前查询的细节； 对于每个查询运行，左上角的文本是查询ID，图三中为：20201116_160228_00002_63fws

前面是YYYYMMDD_HHMMSS格式的日期，具体的时间是当前查询运行时的时间，后半部分是一个自增的计数器，00010的含义表示这个查询时Coordinator重启以来第10个查询，最后的字符：63fws，是随机生成的Coordinator的标识符，每次coordinator重启会重置标识符和计数器。

root：提交该查询的用户

presto-jdbc:查询来源

global:当前查询的Resource Group

最下面的9个指标对应下面的表格；

Completed Splits: 查询的已完成Splits的数目。这个例子显示了25个已完成的Splits。在查询执行的开始时和执行完成时这个值是0。当查询正在进行期间这个值会一直增加 Running Splits: 查询中正在运行的运行Splits的数量。当查询完成时，这个值总是0。但是，在执行过程中，随着Splits的运行和完成，这个数字会发生变化 Queued Splits： 当前查询里出于队列中的Splits数。当查询完成时，这个值总是0。但是，在执行期间，这个数字会发生变化
Wall Time： 执行查询所花费的Wall Time。即使在分页结果时，此值也会继续增长 Total Wall Time: 此值与Wall Time相同，但它也包括排队时间。Wall Time不包括查询排队的任何时间。这是您观察的总时间，从您提交查询到您接收结果 CPU Time： 处理查询所花费的总CPU时间。这个值通常比wallTine时间大，因为如果使用四个CPU花费1秒来处理一个查询，那么总的CPU时间是4秒
Current Total Reserved Memory：当前用于查询执行总的reserved memory使用。对于已完成的查询，此值为0 Peak Total Memory： 查询执行期间的峰值总内存使用量。查询执行期间的某些操作可能需要大量内存，了解峰值是多大是很有用的 Cumulative User Memory： 在整个查询处理过程中使用的累积内存。这并不意味着所有的内存都是同时使用的。它是累积的内存总量

查询明细视图：

通过点击查询ID可以跳转到该查询的明细界面：

image-20201117140000722

Overview页面包括查询列表的查询细节信息如下：

image-20201117140029913

image-20201117140043514

image-20201117140104958

Presto操作

==注：presto连接hive时需要启动hiveserver2和hive元数据服务==

1
2
3
hive --service hiveserver2 &

hive --service metastore &

presto客户端（cmd）

  1. 客户端jar](https://repo1.maven.org/maven2/com/facebook/presto/presto-cli/0.243.2/presto-cli-0.243.2-executable.jar)

  2. 存放至bin目录下，chmod修改可执行权限。

  3. 启动命令：

    1
    ./presto --server 192.168.88.110:8848

    image-20201116152721211

  4. 查看数据源：

    1
    show catalogs;

    image-20201116153040875

  5. 查看mysql中的库

    1
    show schemas from mysql;

    image-20201116153210235

  6. 查询hive库下的某张表中的数据

    1
    select * from hive.db_user.t_farm_day01 limit 10;

    image-20201116153549660

  7. 查询mysql库下的某张表中的数据

    image-20201116153746774

    1
    select * from mysql.test.student;

    image-20201116153825908

javaAPI操作

util

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
//获取连接（硬编码）
public static Connection getConnection(String catalogs,String schemas) throws ClassNotFoundException, SQLException {
       Class.forName("com.facebook.presto.jdbc.PrestoDriver");
       Connection conn = DriverManager.getConnection("jdbc:presto://192.168.88.110:8848/"+catalogs+"/"+schemas, "root", null);
       return conn;
   }
//返回json数据
   public static JSONArray getDataAll(String sql,String catalogs,String schemas) throws Exception {
       Connection con = getConnection(catalogs,schemas);
       JSONArray array = new JSONArray();
       Statement ps = null;
       ResultSet rs = null;

       try {
           ps = con.createStatement();
           rs = ps.executeQuery(sql);
           // 获取列数
           ResultSetMetaData metaData = rs.getMetaData();
           int columnCount = metaData.getColumnCount();
           // 遍历ResultSet中的每条数据
           while (rs.next()) {
               JSONObject jsonObj = new JSONObject();
               // 遍历每一列
               for (int i = 1; i <= columnCount; i++) {
                   String columnName = metaData.getColumnLabel(i);
                   String value = StringUtils.isBlank(rs.getString(columnName)) ? "" : rs.getString(columnName);
                   jsonObj.put(columnName, value);
               }
               array.add(jsonObj);
           }
       } catch (Exception e) {
           throw new Exception("ERROR:" + e.getMessage(), e);
       } finally {
           //关闭资源(先开后关)
           if (null != ps) {
               ps.close();
           }
           if (null != rs) {
               rs.close();
           }
           if (null != con) {
               con.close();
           }
       }
       return array;
   }

查询所有表名

controller

1
2
3
4
5
6
7
8
9
10
@RequestMapping("showTables")
    public List<String> showTables() throws SQLException, ClassNotFoundException {
        String sql = "show tables";
//        String catalogs = "hive";
//        String schemas = "db_user";
        String catalogs = "mysql";
        String schemas = "test";
        List<String> tableNameList = prestoService.showTables(sql,catalogs,schemas);
        return tableNameList;
    }

service

1
2
3
4
5
6
7
8
9
10
11
12
13
public List<String> showTables(String sql,String catalogs,String schemas) throws ClassNotFoundException, SQLException {
    List<String> tableNameList = new ArrayList<>();
    Connection conn = JdbcUtil.getConnection(catalogs,schemas);
    Statement stmt = conn.createStatement();
    ResultSet rs = stmt.executeQuery(sql);
    while (rs.next()) {
        //System.out.println(rs.getString(1));
        tableNameList.add(rs.getString(1));
    }
    rs.close();
    conn.close();
    return tableNameList;
}

hive表：

image-20201117083942103

mysql表：

image-20201117110743857

查询hive库下某张表中的数据

1
select * from hive.db_user.t_farm_day01 limit 10;

controller

1
2
3
4
5
6
@RequestMapping("select")
    public JSONArray select() throws Exception {
        String sql = "select * from t_farm_day01 limit 10";
        JSONArray jsonArray = prestoService.select(sql);
        return jsonArray;
    }

service

1
2
3
4
5
6
7
8
9
10
11
12
/**
    *
    * @param sql
    * @param catalogs  数据库类型（mysql hive pg）
    * @param schemas  具体数据库类型下的数据库
    * @return  json数组
    * @throws Exception
    */
   public JSONArray select(String sql,String catalogs,String schemas) throws Exception {
       JSONArray array = JdbcUtil.getDataAll(sql,catalogs,schemas);
       return array;
   }

image-20201117105203028

适用于Tableau的Web连接器

用于Tableau的Presto Web连接器使用户可以针对Presto从Tableau运行查询。

在Spark上执行Presto

Presto on Spark使利用Spark作为Presto查询的执行框架成为可能。这对于我们要在数千个节点上运行，需要10或100 TB的内存以及消耗很多CPU年的查询很有用。

架构部分摘自知乎Presto详解

您的支持是我继续创作的动力