Hive SQL 迁移的动机

Flink 已经是流计算的事实标准，当前国内外做实时计算或流计算一般都会选择 Flink 和 Flink SQL。另外，Flink 也是是家喻户晓的流批一体大数据计算引擎。

然而，目前 Flink 也面临着挑战。比如虽然现在大规模应用都以流计算为主，但 Flink 批计算的应用并不广泛，想要进一步推动真正意义上的流批一体落地，需要推动业界更多地落地 Flink 批计算，需要更积极地拥抱现有的离线生态。当前业界离线生态主要以 Hive 为主，因此我们在过去版本中做了很多与 Hive 相关的集成，包括 Hive Catalog、Hive 语法兼容、Hive UDF 兼容、流式写入 Hive 等。在 Flink 1.16 版本中，我们进一步提升了 HiveSQL 的兼容度，还支持了 HiveServer2 的协议兼容。

所以，为什么 Flink 要去支持 Hive SQL 的迁移？一方面，我们希望吸引更多的 Hive 离线数仓用户，通过用户来不断打磨批计算引擎，对齐主流批计算引擎。另一方面，通过兼容 Hive SQL，来降低现有离线用户使用 Flink 开发离线业务的门槛。除此之外，另外，生态是开源产品的最大门槛。Flink 已经拥有非常丰富的实时生态工具，但离线生态依然较为欠缺。通过兼容 Hive 生态可以快速融入 Hive 离线生态工具和平台，降低用户接入的成本。最后，这也是实现流批一体的重要一环，我们希望推动业界尝试统一的流计算和批计算引擎，再统一流计算和批计算 SQL。

从用户角度来看，Hive SQL 为什么要迁移到 Flink SQL 上？

对于平台方而言，统一流批计算引擎，只需维护一套 Flink 引擎，可以降低维护成本，提升团队研发效率。另外，可以利用 Flink + Gateway+ HiveSQL 兼容，快速建设一套 OLAP 系统。Flink 的另一优势是拥有丰富的 connector 生态，可以借助 Flink 丰富的数据源实现强大的联邦查询。比如不仅可以在 Hive 数仓里做 ad-hoc 查询，也可以将 Hive 表数据与 MySQL、HBase、Iceberg、Hudi 等数据源做联邦查询等。

对于离线数仓用户而言，可以用 Hive SQL 写流计算作业，极大降低实时化改造成本。使用的依然是以前的 HiveSQL 语法，但是可以运行在 streaming 模式下。在此基础之上也可以进一步探索流批一体 SQL 层以及流批一体数仓层的建设。

Hive SQL 迁移的挑战

但是 Flink 支持 HiveSQL 的迁移面临着很多挑战，主要有以下三个方面：

• 兼容：包括离线数仓作业和Hive平台工具的兼容。主要对应用户层的兼容和平台方的兼容。
• 稳定性：迁移后的作业首先要保证生产的稳定性。我们在1.16中也做了很多这方面的工作，包括FLIP-168 预测执行和Adaptive Hash Join。后续我们会发表更多的文章来介绍这方面的工作。
• 性能：最后性能也是很重要的，在1.16中我们也做了很多这方面的工作，包括Dynamic Partition Pruning（DPP）、元数据访问加速等，后续也会发表更多文章来介绍这方面的工作。

接下来我们重点讲解下 Hive 兼容相关的工作。

Hive 语法的兼容并没有完全造出一套新的 SQL 引擎，而是复用了 Flink SQL 的很多核心流程和代码。我们抽象出了可插拔的 parser 层来支持和扩展不同的语法。Flink SQL 会经过 Flink Parser 转换成 Flink RelNode，再经过 Logical Plan 优化为 Physical Plan，最后转换为 Job Graph 提交执行。为了支持 Hive 语法兼容，我们引入了 Hive Parser 组件，来将 Hive SQL 转化成 Flink RelNode。这个过程中，复用了大部分 Hive 现有的 SQL 解析逻辑，保证语法层的兼容（均基于 Calcite）。之后 RelNode 复用同样的流程和代码转化成 LogicalPlan、Physical Plan、JobGraph，最后提交执行。

从架构上看，Hive 语法兼容并不复杂，但这是一个“魔鬼在细节”的工作。上图为部分 Flink1.16 版本里 Flink Hive 兼容相关的 issue，涉及 query 兼容、类型系统、语义、行为、DDL、DML、辅助查询命令等非常多语法功能。累计完成的 issue 数达近百个。

Flink1.16 版本将 Hive 兼容度从 85% 提升至 94.1%。兼容度测试主要依靠 Hive qtest 测试集，其中包含 12,000 多个测试 case，覆盖了 Hive 目前所有主流语法功能。没有兼容的一部分包括 ACID 功能（业界使用较少），如果除去 ACID 功能，兼容度已达 97%以上。

SQLGateway 是 Flink SQL 的 server 层组件，是单独的进程，对标 HiveServer2 组件。从 Flink 整体架构上看，SQLGateway 处于中间位置。

向下，封装了用户 API 的 Flink SQL 和 Hive SQL。不管是 Flink SQL 还是 Hive SQL，都使用 Flink 流批一体的 Runtime 来执行，可以运行在批模式，也可以运行在流模式。Flink 的资源也可以部署运行在 YARN、K8S、Flink standalone 集群上。

向上，SQLGateway 提供了可插拔协议层 Endpoint，目前提供了 HiveServer2 和 REST 两种协议实现。通过 HiveServer2 Endpoint，用户可以将 Hive 生态的很多工具和组件（Zeppelin、Superset、Beeline、DBeaver 等）连接到 SQL Gateway，提供流批统一的 SQL 服务并兼容 Hive SQL。通过 REST 协议可以使用 Postman、curl 命令或自己通过 Python、Java 编程来访问，提供完善和灵活的流计算服务。将来，Endpoint 能力也会继续扩展，比如可以提供更高性能的 gRPC 协议或兼容 PG 协议。

Hive SQL 迁移的实践

目前快手正在与 Flink 社区紧密合作，推进流批一体的落地。目前快手迁移 Hive SQL 作业到 Flink SQL 作业已经取得了初步的进展，已有上千个作业完成了迁移。快手的迁移主要策略为双跑平台，已有业务继续运行，双跑平台有智能路由组件，可以通过指定规则或 pattern 识别出作业，投递到 MapReduce、Spark 或 Flink 上运行。初期的运行较为谨慎，会通过白名单机制指定某些作业先运行在 Flink，观察其稳定性与性能，对比其结果一致性，后续逐步通过规则来放量。更多的实践经验与细节可以关注 Flink Forward Asia 2022 上分享的《Hive SQL 迁移到 Flink SQL 在快手的实践》。

Hive SQL 迁移的演示

Demo：Hive SQL 如何迁移到 Flink SQL

接下来演示一下 Hive SQL 如何迁移到 Flink SQL。我们已经搭建好一个 YARN 集群，以及 Hive 相关组件，包括 HiveServer2 的服务。我们使用 Zeppelin 做数据可视化和 SQL 查询。我们将演示 Hive SQL 迁移到 Flink SQL 只需改一行地址，Zeppelin 体验并无二致，SQL 也无需修改。完整的 Demo 视频请观看完整的演讲视频：https://www.bilibili.com/video/BV1BV4y1T7d4

未来规划

未来，Flink 将在以下三个方面持续演进：

• 第一，持续在 batch 上做更多尝试和投入，提升 batch 的稳定性和性能，目标是短期内能够追齐主流的批计算引擎。
• 第二，完善数据湖的分析，比如更高效的批式数据湖读写、查询优化下推、列存上的读写优化，Iceberg、Hudi 以及 Flink Table Store 的支持等。另外，也会提供丰富的湖上数据查询和管理功能，比如查询快照版本的能力、查询元数据、更丰富的 DML 语法（UPDATE、DELETE、MERGE INTO）以及管理湖上数据 CALL 命令等。
• 第三，Flink Batch 生态建设，包括进一步完善 Remote Shuffle Service、血缘管理等。