Presto/Trino分布式联邦查询引擎

可视化演示

展示Trino Coordinator+Workers MPP架构（SQL→解析→计划→调度→并行执行→汇总）、跨数据源联邦查询（Hive+MySQL+Kafka→Connector→JOIN→统一结果）与性能对比（MPP并行加速比+Trino vs Spark SQL延迟/吞吐/并发/容错），交互切换核心维度。

架构原理：中央调度与分片执行

美食广场里，二十个档口各卖各的，你要一份“杭州订单 + 用户信息”的汇总，不可能每个档口都去问一遍。广场中央设了一个调度台：它先收下订单，扫一眼菜单，把任务拆成几张便签，交给几个跑腿小哥分头行动——有人去烤肉档取用户数据，有人去拉面档取订单数据，最后在调度台拼好菜端给你。整个过程，你只需要下这一次单。

Presto 就是这样一座“美食广场”级的分布式 SQL 查询引擎，但它自己不存储任何数据。集群里有两类角色：Coordinator（协调者）就是调度台，负责接收 SQL、解析成执行计划、把任务分发给 Worker；Worker（工作者）就是跑腿小哥，负责并行读取数据源中的分片（Split），完成过滤、聚合、连接等计算。Presto 后因商标问题改名为 Trino，社区主线延续，但大家习惯上仍称 Presto。

一条查询的生命周期大致是：客户端把 SQL 发给 Coordinator → Coordinator 经过词法/语法解析生成逻辑执行计划，再优化为物理执行计划 → 把计划切成多个可并行执行的 Task → 通过服务发现机制找到空闲 Worker 分发任务 → Worker 通过 Connector 从各数据源拉取数据并计算 → 结果以流水线方式回流汇总 → 返回给客户端。这种无共享架构让 Presto 可以线性扩展：加 Worker，查询更快。

所以，你对美食广场的感知只有一句“下单→上菜”，但调度台和小哥们在背后完成了一整出分头行动的交响。这正是 Presto 的设计哲学：对外是极简的 SQL 接口，对内是精巧的分布式流水协作。

Connector机制：连通一切数据源

每个档口的点餐方式都不一样：烤肉档要求报号，拉面档要手写单，甜品档只认扫码。调度台总不能为每个档口记一套规矩。于是每个档口门口挂了一个“点餐翻译器”：你按统一菜单点，翻译器把标准订单翻译成档口认得的语言，再把档口端出的菜品翻译成统一格式。

这个“点餐翻译器”就是 Presto 的 Connector（连接器）。它是连接 Presto 与具体数据源的插件化适配层，定义了统一的 SPI（服务提供接口）。目前社区有 50 多个官方和第三方 Connector，覆盖 MySQL、PostgreSQL、Hive、HDFS、Kafka、MongoDB、Elasticsearch 甚至远程 HTTP API。只要实现一套 SPI，任何数据源都能被 SQL 查询。

Connector 的核心抽象有三块：TableLayout（表布局，描述表如何分区、分桶）、Split（数据分片，描述一段可独立读取的数据范围）、PageSource（数据读取器，真正拉数据并转成 Presto 统一的行列格式）。查询时，Connector 的 getSplits 会把一张表按分区、文件块等拆成多个分片，Worker 并行拉取，互不干扰。

有了翻译器，无论档口背后是烤肉炉还是数据库，调度台看到的都是同样的标准菜单。Connector 让 Presto 的联邦查询能力变得“即插即用”：想接入新数据源？写个翻译器挂上去就行。

查询优化器：聪明的执行规划

一个经验丰富的调度员接到订单后，不会拔腿就跑。他会先盘算：客人要“杭州用户”的姓名和金额，那就让 MySQL 档口先把杭州的筛出来，只取两个字段，别把整本账本搬回来；两个档口的食材怎么合并最省事，是把小份的用户表搬到大份订单那边，还是两边都按用户编号分桶再拼。这些盘算，就是优化器。

Presto 的查询优化器基于关系代数做等价变换，目标是生成“最便宜”的物理执行计划。最核心的手段有两个：谓词下推和列裁剪。谓词下推把 WHERE 过滤条件尽可能下推到数据源或最底层算子执行，让数据在源头就被筛掉一部分；列裁剪则让每个算子只读取 SELECT 涉及的列，大幅减少 IO。此外还有 Join 重排：根据表大小选择 Broadcast Join（小表广播）或 Partitioned Join（分区匹配）。

CBO（基于成本的优化器）则进一步让选择变得“有据可依”：它依据表的统计信息（行数、字段基数、数据分布），为每个候选算子估算 CPU 开销与 IO 开销，求和得到整体成本，最终选取成本最小的执行计划。优化器会尝试多种等价执行计划，如同调度员在脑中模拟多条取餐路线。

一个聪明调度员的经验，被编码成了可计算的成本公式。优化器能自动替成千上万的查询“想清楚”，这也是为什么同样一条 SQL，Presto 常常比人肉写的流程快一个数量级。

内存管理：取餐台的生存法则

档口做好的菜不能直接砸到客人脸上，得放到取餐台暂存。取餐台就那么大，订单太多放不下怎么办？要么把菜直接倒掉（查询失败），要么把半成品搬到后面的储藏室（磁盘），等取餐台空出来再挪回来。Presto 选择了后者，这就是内存溢写（Spill）。

Presto 是内存优先的纯计算引擎，Worker 的内存被划分为多个池：系统内存池（System Pool）给调度和内部服务用；用户内存池（User Pool）存聚合、过滤、连接的中间结果；保留池（Reserved Pool）为突发大查询兜底。每个查询在单节点上有内存上限，超过即触发溢写或失败，避免一个查询吃光整个节点内存拖垮其他查询。

开启 spill 后，当 join、聚合等算子的中间结果超过内存阈值，Presto 会把溢出部分按顺序写到本地磁盘，需要时再读回来。虽然磁盘交换会变慢，但总比整个查询 OOM 崩掉好。核心参数有三个：query.max-memory-per-node（单查询单节点内存上限）、query.max-total-memory-per-node（含系统开销的总上限）、spill-enabled（是否开启溢写）。

取餐台的管理哲学是：宁可让部分菜品暂存储藏室，也不让整个美食广场因堆满菜品而瘫痪。合理配置内存池和溢写阈值，Presto 才能在高并发下既不 OOM 也不频繁落盘。

与Spark对比：美食广场与中央大厨房

如果美食广场是“现点现做”，那还有一种“中央大厨房”模式：食材提前集中拉进厨房，一次性做成几千份盒饭，准备时间长但吞吐惊人。Presto 是前者，Spark SQL 是后者。你很难说谁更好，只能说谁更适合当下的订单——是等不及的单客，还是大批量团餐客户。

Presto 是 MPP（大规模并行处理）无状态查询引擎，Coordinator 分片调度、Worker 内存流水线计算，目标是把查询延迟压到秒级。Spark SQL 是批处理框架，基于 RDD/Dataset 的弹性调度，把一个任务切成多个 Stage 和 Task 迭代执行，牺牲一定的调度开销换取容错和稳定性，适合分钟级以上的大规模 ETL。

两者的差异落在地板与天花板：Presto 在 GB 到 TB 级交互查询上通常比 Spark 快 5 到 10 倍；但面对 PB 级数据、复杂多轮迭代、机器学习训练，Spark 的稳定性和生态更胜一筹。Presto 也能跑 INSERT INTO 做轻量 ETL，但复杂的数仓加工仍然是 Spark 的主场。选型时重点看三个维度：延迟要求、数据规模、计算复杂度。

美食广场和中央大厨房会长期共存：想吃一份热乎的餐前小点，找 Presto；要做三千份团餐，找 Spark。理解两者分工，才不会被“Presto 比 Spark 快”这种片面结论带偏。

性能调优：让每一单更快

排队高峰期，怎么让每个客人都更快拿到菜？多开几个取餐窗口（提高并行度），把爆款菜提前备好（缓存热数据），精确记下客人点的菜而不是“每样来一份”（分区与列裁剪），把顺路的订单合并配送（Join 优化）。美食广场的这些招数，Presto 也一一对应。

Presto 调优的第一步是看并行度。并行度受限于两个因素：总分片数 S 和 Worker 计算能力（核数 × Worker 数）。分片太少则 Worker 闲置，太多则调度开销上升，经验上让每个 Worker 处理的 split 数在 2 到 4 个。第二步是利用数据源的分区信息做分区裁剪：Hive 分区表只扫描命中分区，能省掉 90% 以上的 IO。

第三步是 Join 调优：小表 join 大表时把 join_distribution_type 设为 BROADCAST，小表广播到所有 Worker，避免大 Shuffle；大表 join 大表则用分区 Join，并根据 Join Key 的分布倾斜情况调整分区数。最后，热数据可以挂在分布式缓存层（如 Alluxio）或开启结果缓存，减少重复拉取底层存储的压力。

调优的本质是找到瓶颈在哪一个环节：是档口出菜慢（数据源 IO），还是跑腿小哥不够（并行度），还是取餐台太挤（内存）。对症下药，比盲目加机器更有效。

资源隔离：多租户的秩序管理

美食广场最怕什么？一个网红档口排了一百号人，把通道堵死，其他客人全进不来。物业于是定规矩：每个档口最多同时接待几单，超出就排队；VIP 团餐可以插队，普通小单按顺序来。这样谁都挤不垮谁。

Presto 用“资源组”（Resource Group）实现多租户隔离。你可以在配置里定义组名、权重、并发上限和排队策略，再用正则或会话属性把不同业务的查询路由到对应组。比如 BI 组权重 10，临时查询组权重 2，高峰期临时大查询进队列，不能抢占 BI 的查询资源。

每个资源组有四个核心参数：hardConcurrencyLimit（组内最大并发运行数）、maxQueued（最大排队数）、maxRunning（含排队的最大总数）、schedulingWeight（调度权重，决定轮转分配时的资源比例）。调度器按权重在组间轮转分配槽位，权重高的组拿到更多并发。联邦查询集群往往被多个团队共享，没有资源组，一个跑偏的大查询就能拖垮全公司报表。

资源隔离不是简单的限流，而是“秩序”：让重要查询有保障，让临时任务不饿死，让集群在洪峰下依旧稳定。这也是 Presto 上生产环境的关键一步。

AI 应用场景

Python 演示

import time # 模拟两个异构数据源: mysql_users 类似 MySQL 表, hdfs_orders 类似 HDFS 文件mysql_users = [    {"id": 1, "name": "Alice", "city": "杭州"},    {"id": 2, "name": "Bob", "city": "上海"},    {"id": 3, "name": "Carol", "city": "杭州"},] hdfs_orders = [    {"user_id": 1, "amount": 100},    {"user_id": 1, "amount": 200},    {"user_id": 2, "amount": 50},] def scan_with_predicate(data, column, value):    # 模拟谓词下推: 在数据源端过滤, 而不是把全表拉回内存    return [row for row in data if row.get(column) == value] def federated_join(user_table, order_table, user_key, order_key, city_filter=None):    # 第1步: 谓词下推, 在 MySQL 侧先过滤城市    if city_filter:        user_table = scan_with_predicate(user_table, "city", city_filter)    # 第2步: 构造 hash 索引, 模拟把小表广播到各分片    user_index = {row[user_key]: row for row in user_table}    # 第3步: 遍历订单分片, 与用户索引做 join    rows = []    for order in order_table:        user = user_index.get(order[order_key])        if user:            rows.append({**user, **order})    return rows if __name__ == "__main__":    start = time.time()    result = federated_join(mysql_users, hdfs_orders, "id", "user_id", city_filter="杭州")    cost_ms = (time.time() - start) * 1000    print("联邦查询结果:", result)    print("耗时: %.2f ms" % cost_ms)

1. def scan_with_predicate(data, column, value):定义了过滤函数，模拟数据源侧的谓词下推，让数据在源头先变少。
2. return [row for row in data if row.get(column) == value]列表推导式逐行过滤，只保留城市匹配的用户记录。
3. def federated_join(...)联邦查询主函数，接收两个“数据源”和关联键，模拟一条跨源 SQL 的执行过程。
4. user_table = scan_with_predicate(user_table, "city", city_filter)先把杭州用户筛出来，相当于 WHERE city='杭州' 被推到了 MySQL 端执行。
5. user_index = {row[user_key]: row for row in user_table}构建哈希索引，模拟 Presto 把小表广播到各分片后建立的内存查找表。
6. user = user_index.get(order[order_key])扫描订单“分片”时，用 O(1) 哈希查找匹配用户，避免笛卡尔积式全表扫描。
7. rows.append({**user, **order})将用户和订单字段拼接合并，模拟联邦查询最终把多源结果组装成宽表返回。

常见误区

面试题