今天聊聊统一 SQL 网关：Apache Kyuubi。他的核心在于能够支持多租户和分布式，允许用户通过 JDBC、 SQL 客户端或者 HTTP 提交 SQL 查询到不同的数据处理引擎，比如 Spark、Flink、Hive、Trino、Jdbc。
本文将介绍 Kyuubi 的发展历程，架构设计，主要特性，应用场景，安装部署，各种引擎的交互式查询操作以及批处理操作。

Kyuubi是什么？

简介

Apache Kyuubi（Incubating）是一个分布式 和 多租户网关，Kyuubi 在各种现代计算框架之上构建分布式 SQL 查询引擎，例如 Apache Spark、 Flink、 Doris、 Hive、 Trino 和 StarRocks 等，以查询来自异构数据源的分布在机器群上的海量数据集。

发展历程

生态

下方的 Kyuubi Server 泳道图将我们的潜在用户分为最终用户和管理员。一方面，它向最终用户隐藏了计算和存储的技术细节，因此，他们可以使用熟悉的工具专注于业务和数据；另一方面，它向管理员隐藏了业务逻辑的复杂性，因此，他们可以在零维护停机时间内升级服务器端的组件，在清晰了解最终用户正在做什么的情况下优化工作负载，确保集群和数据安全的身份验证、授权和审计等等。

以上就是 Kyuubi 的完整生态，每层都松散地耦合到其他层。我们可以将上面的一些组件组合起来构建现代数据栈，比如可以使用 Kyuubi、Spark 和 Iceberg 构建和管理纯 SQL 的数据仓库，用于 ETL 和 OLAP，所有的工作都可以在一个平台上完成，使用一份数据副本和一个 SQL 接口。

Kyuubi 的架构设计

Kyuubi 的架构也非常简单，包括 Client、Server、Engine，如下所示：

Kyuubi Client
接受用户提交的 SQL，并提交给 Server。
Kyuubi Server
管理 Session、Operation、Engine，接受 Client 的请求，并交给 Engine 执行。
Kyuubi Engine
实际的计算引擎，比如 Flink、Spark、Trino、Hive。

在 Kyuubi 架构中 Client 与 Server、Server 与 Engine 之间的通信都是基于 Zookeeper 的服务发现和负载均衡，这意味着不同模块之间是非常松耦合的。在同一个 Zookeeper namespace 下，Client 可以连到任意一个 Server 上， Server 也能访问任意一个 Engine。

一个请求最终使用哪个 Engine 执行由 Kyuubi Server 自动控制。Kyuubi Server 会根据多租户隔离策略的自动选择一个合适的 Engine，如果满足要求的 Engine 不存在，就会即时启动一个。当一个 Engine 闲置一段时间后，Engine 会自动退出来节省资源，当然也可以通过 Kyuubi Server 主动关闭 Engine。可以说 Kyuubi Engine 是按需启动的，生命周期也是由 Server 管理的。

而什么时候启动 Engine 是由引擎的隔离级别决定的，Kyuubi 提供 4 种引擎隔离级别：

CONNECTION
CONNECTION 级别意味着每个连接独占一个 Engine，这是最高的隔离级别，比较适合离线的大作业、Ad-hoc 查询或者实时作业的情况。
USER
USER 级别意味着每个用户独占一个 Engine，适合大部分场景。
GROUP
GROUP 级别则是每个用户组共享一个 Engine，适合小数据量的场景。
SERVER
SERVER 级别则是全部用户共享全局的一个 Engine，这是最低的隔离级别，一般是管理员执行管理操作时使用。

主要特性

多租户
Kyuubi 通过统一的身份验证授权层为资源获取和数据 / 元数据访问提供端到端的多租户支持。
高可用
Kyuubi 通过 ZooKeeper 提供负载均衡，它提供了企业级的高可用性，以及无限的客户端高并发。
多个工作负载
Kyuubi 可以通过一个平台、一个数据副本和一个 SQL 接口轻松支持多个不同的工作负载。
统一接口
Kyuubi 可以与多种大数据生态系统集成，并提供了标准的 JDBC 接口，如 Spark、Flink、Hive、Trino、Jdbc 等，满足不同场景的需求。
高性能
Kyuubi 针对大数据场景进行了优化，支持谓词下推、查询缓存等特性，提高查询性能。
完善的监控指标
可以通过指标、日志等查看 kyuubi 服务器和引擎的故障和系统负载。
提供多种接口协议
Hive Thrift 协议、Restful Api、MySQL 协议

应用场景

交互式分析
Kyuubi 是一个先进的企业级快速分析平台，用于对大数据进行交互式可视化分析，支持常见的计算框架，即 Apache Spark、Apache Flink、Trino 等。

使用 JDBC/ODBC，用户可以直接通过 SQL 或通过 BI 工具来访问 kyuubi 并高效地运行查询。

Kyuubi 在用户级别缓存后台引擎实例，以更好的实现计算资源共享和快速响应，它们能并行处理大量数据的查询并快速返回结果。

批量处理
Kyuubi 提供了一个 SQL 接口用于常见的批处理，通常是大型提取、转换、加载 (ETL) 过程。
Kyuubi 及其引擎都是存储独立的，支持众多的数据源，并且 Kyuubi 在连接级别隔离了后台引擎实例，以便实现更好的计算资源隔离和稳定性。
数据湖和 Lakehouses
Kyuubi 支持同时查询所有传统数据仓库，如 Apache Hive/HDFS，或现代的 Lakehouse，如 Apache Iceberg、Apache Hudi 和 Delta Lake。

Kyuubi 还提供多目录元数据 API，可在用户面前呈现所有数据的大致轮廓，并帮助用户更快地进行创新。使用 ANSI 标准 SQL 语法在单个入口中查询不同数据源的能力极大地简化了数据洞察，同时提供身份验证和授权以确保所有数据的安全。

安装部署

环境准备

组件	角色	版本	备注
Java	jre	8、11、17
Kyuubi	网关引擎	1.9.1	Kyuubi Server、Kyuubi Engine、Kyuubi Hive Beeline
Spark	Engine		Spark 环境
Flink	Engine		Flink 环境
Trino	Engine		Trino 环境
Doris	Engine	1.2.8	Doris 环境
Hive	Engine Metastore	2.1、2.3、3.1	Hive 环境
Zookeeper	HA

下载 Kyuubi

下载最新版的 Kyuubi：apache-kyuubi-1.9.1-bin.tgz，解压放到 /opt/software, 并重命名为：kyuubi-1.9.1

配置 Kyuubi

配置文件在安装目录的conf中。

kyuubi-env.sh

# java 环境
export JAVA_HOME=/opt/software/jdk1.8.0_381
# Spark 环境
export SPARK_HOME=/opt/software/spark-3.4.2-bin-hadoop3
# Flink 环境
export FLINK_HOME=/opt/software/flink-1.16.2
export FLINK_HADOOP_CLASSPATH=`hadoop classpath`
# Hive 环境
export HIVE_HADOOP_CLASSPATH=${HADOOP_HOME}/share/hadoop/common/lib/commons-collections-3.2.2.jar:${HADOOP_HOME}/share/hadoop/client/hadoop-client-runtime-3.3.6.jar:${HADOOP_HOME}/share/hadoop/client/hadoop-client-api-3.3.6.jar:${HADOOP_HOME}/share/hadoop/common/lib/htrace-core4-4.1.0-incubating.jar

kyuubi-defaults.conf

# kyuubi高可用属性
kyuubi.ha.enabled=true
kyuubi.ha.zookeeper.quorum=hadoop1,hadoop2,hadoop3
kyuubi.ha.zookeeper.client.port=2181
kyuubi.ha.zookeeper.session.timeout=600000

# 默认查询引擎
kyuubi.engine.type=SPARK_SQL

启动

bin/kyuubi start

kyuubi 还贴心的为我们提供了WEB UI，可以通过 WEB 页面可视化管理我们的各种组件和引擎，同时还可以在线提交 sql，不过目前还不支持引擎的选择，只支持配置文件中配置的默认引擎。kyuubi 启动之后，通过以下地址就可以访问 web 页面了：http://hadoop1:10099

交互式查询

Beeline方式

# 使用 Spark 引擎
bin/beeline -u 'jdbc:hive2://hadoop1:10009/;#kyuubi.engine.type=SPARK_SQL;' -n apache

# 使用 Hive 引擎
bin/beeline -u 'jdbc:hive2://hadoop1:10009/;#kyuubi.engine.type=HIVE_SQL;' -n apache

# 使用 Flink 引擎
bin/beeline -u 'jdbc:hive2://hadoop1:10009/;#kyuubi.engine.type=FLINK_SQL;' -n apache

# 使用 Trino 引擎
bin/beeline -u 'jdbc:hive2://hadoop1:10009/;#kyuubi.engine.type=TRINO;kyuubi.session.engine.trino.connection.url=http://hadoop1:18080;kyuubi.session.engine.trino.connection.catalog=hive' -n apache

# 使用 jdbc 引擎
bin/beeline -u 'jdbc:hive2://hadoop1:10009/;#kyuubi.engine.type=jdbc;kyuubi.engine.jdbc.type=mysql;kyuubi.engine.jdbc.driver.class=com.mysql.cj.jdbc.Driver;kyuubi.engine.jdbc.connection.url=jdbc:mysql://hadoop1:3306/test;kyuubi.engine.jdbc.connection.user=demo;kyuubi.engine.jdbc.connection.password=123456;kyuubi.session.user=test;' -n apache

Jdbc方式

① 引入 kyuubi jdbc 依赖

<dependency>
    <groupId>org.apache.kyuubigroupId>
    <artifactId>kyuubi-hive-jdbc-shadedartifactId>
    <version>1.7.0version>
dependency>

② demo

import java.sql.*;

public class KyuubiJDBC {

    public static void main(String[] args) throws SQLException {
        String jdbcUrl = "jdbc:kyuubi://hadoop1:10009/;#kyuubi.engine.type=jdbc;" +
                "kyuubi.engine.jdbc.type=mysql;" +
                "kyuubi.engine.jdbc.driver.class=com.mysql.cj.jdbc.Driver;" +
                "kyuubi.engine.jdbc.connection.url=jdbc:mysql://172.16.143.1:3306/test;" +
                "kyuubi.engine.jdbc.connection.user=shengyu;" +
                "kyuubi.engine.jdbc.connection.password=123456;" +
                "kyuubi.session.user=test;";

        String spark_sql_url = "jdbc:kyuubi://hadoop1:10009/;#kyuubi.engine.type=SPARK_SQL;";
        String hive_sql_url = "jdbc:kyuubi://hadoop1:10009/;#kyuubi.engine.type=HIVE_SQL;";
        String flink_sql_url = "jdbc:kyuubi://hadoop1:10009/;#kyuubi.engine.type=FLINK_SQL;";
        String trino_url = "jdbc:kyuubi://hadoop1:10009/;#kyuubi.engine.type=TRINO;" +
                "kyuubi.session.engine.trino.connection.url=http://hadoop1:18080;" +
                "kyuubi.session.engine.trino.connection.catalog=hive";
//        testEngine(jdbcUrl,"select 1");
//        testEngine(spark_sql_url,"select 1");
//        testEngine(hive_sql_url,"select 1");
//        testEngine(flink_sql_url,"select 1");
        testEngine(trino_url, "select * from hive.sjzt.\"5g_http\" limit 10");
    }


    public static void testEngine(String kyuubi_url, String sql) throws SQLException {
        try (Connection conn = DriverManager.getConnection(kyuubi_url)) {
            try (Statement stmt = conn.createStatement()) {
                try (ResultSet rs = stmt.executeQuery(sql)) {
                    while (rs.next()) {
                        System.out.println(rs.getString(1));
                    }
                }
            }
        }
    }
}

Http方式

① 创建 session

# 创建 jdbc session，其他引擎类似
POST http://hadoop1:10099/api/v1/sessions
{
    "configs":{
        "kyuubi.engine.type": "jdbc",
        "kyuubi.engine.jdbc.type":"mysql",
        "kyuubi.engine.jdbc.driver.class":"com.mysql.cj.jdbc.Driver",
        "kyuubi.engine.jdbc.connection.url":"jdbc:mysql://hadoop1:3306/test",
        "kyuubi.engine.jdbc.connection.user":"demo",
        "kyuubi.engine.jdbc.connection.password":"123456",
        "kyuubi.session.user":"demo"
    }
}

response：
{
    "identifier": "05d5708d-d31c-4014-9e15-597fe9570625",
    "kyuubiInstance": "hadoop1:10099"
}

② 执行 sql 查询

POST http://hadoop1:10099/api/v1/sessions/05d5708d-d31c-4014-9e15-597fe9570625/operations/statement
{
    "runAsync":"false",
    "statement":"select 1"
}
response：
{
    "identifier":"fafbe0b3-4f3a-4100-900e-9129139b5dcc"
}

③ 获取 sql 查询结果

GET http://hadoop1:10099/api/v1/operations/fafbe0b3-4f3a-4100-900e-9129139b5dcc/rowset?operationHandleStr=05d5708d-d31c-4014-9e15-597fe9570625&fetchorientation=FETCH_NEXT&maxrows=10

④ 获取查询元数据

GET http://hadoop1:10099/api/v1/operations/fafbe0b3-4f3a-4100-900e-9129139b5dcc/resultsetmetadata?operationHandleStr=05d5708d-d31c-4014-9e15-597fe9570625

⑤ 获取查询日志

GET http://hadoop1:10099/api/v1/operations/fafbe0b3-4f3a-4100-900e-9129139b5dcc/log

⑥ 关闭查询

DELETE http://hadoop1:10099/api/v1/admin/operations/fafbe0b3-4f3a-4100-900e-9129139b5dcc

⑦ 关闭session

DELETE http://hadoop1:10099/api/v1/sessions/05d5708d-d31c-4014-9e15-597fe9570625

⑧ 查询执行状态

curl http://hadoop1:10099/api/v1/operations/faf23e7c-6f5d-4652-8307-abfc133c1a9b/event

⑨ 取消查询

PUT http://hadoop1:10099/api/v1/operations/e32544f4-27bf-4464-8644-a9ac04b0fa81
{
    "action":"cancel"
}

批处理接口

Kyuubi 除了让用户可以使用 HTTP 提交 SQL 查询到不同的引擎，还提供了批处理接口，让用户可以使用 Rest Api 方式提交自定义的 Spark jar 包来实现批处理任务，以下是几个常用的批处理接口：

① 创建批处理session

curl --location --request POST 'http://hadoop1:10099/api/v1/batches' \
--form 'batchRequest="{\"batchType\":\"SPARK\",\"className\":\"org.apache.spark.examples.SparkPi\",\"name\":\"Spark Pi\"}";type=application/json' \
--form 'resourceFile=@"/opt/software/spark-3.4.2-bin-hadoop3/examples/jars/spark-examples_2.12-3.4.2.jar"'

② 查询结果

curl http://hadoop1:10099/api/v1/batches/019b8e91-deee-4432-a162-fcd928a4230a

③ 查询日志

curl http://hadoop1:10099/api/v1/batches/019b8e91-deee-4432-a162-fcd928a4230a/localLog

④ 关闭本次查询

delete http://hadoop1:10099/api/v1/batches/019b8e91-deee-4432-a162-fcd928a4230a

⑤ 查看所有批处理任务

curl http://hadoop1:10099/api/v1/batches

总结

通过以上分享，我们发现 Kyuubi 可以应用于交互式查询和批处理场景中，并且兼容多种查询引擎，比如：Spark、Flink、Hive、Trino、JDBC，我们可以根据实际业务灵活的路由到不同的查询引擎上。而且结合我们实际的元数据管理系统，完全可以屏蔽底层查询引擎的差异和数据位置的差异，从而让数据分析人员和业务人员能够更轻松的应对大数据场景的即席查询，让数据探索变得更加容易。

往期推荐

Trino 动态Catalog 体验

关于clickhouse的几点总结和建议

ClickHouse UDF 实践

看完这篇，你就知道怎么选用ClickHouse表引擎了

Doris Catalog入门学习

统一SQL网关：Kyuubi 使用体验