Flink SQL你用了吗？

news2024/9/26 1:20:22

分析&回答

Flink 1.1.0：第一次引入 SQL 模块，并且提供 TableAPI，当然，这时候的功能还非常有限。
Flink 1.3.0：在 Streaming SQL 上支持了 Retractions，显著提高了 Streaming SQL 的易用性，使得 Flink SQL 支持了复杂的 Unbounded 聚合连接。
Flink 1.5.0：SQL Client 的引入，标志着 Flink SQL 开始提供纯 SQL 文本。
Flink 1.9.0：抽象了 Table 的 Planner 接口，引入了单独的 Blink Table 模块。Blink Table 模块是阿里巴巴内部的 SQL 层版本，不仅在结构上有重大变更，在功能特性上也更加强大和完善。
Flink 1.10.0：作为第一个 Blink 基本完成 merge 的版本，修复了大量遗留的问题，并给 DDL 带来了 Watermark 的语法，也给 Batch SQL 带来了完整的 TPC-DS 支持和高效的性能。

CDC 支持

SQL 1.11 Flink SQL 在原有的基础上扩展了新场景的支持：

Flink SQL 引入了对 CDC（Change Data Capture，变动数据捕获）的支持，它使 Flink 可以方便地通过像 Debezium 这类工具来翻译和消费数据库的变动日志。
Flink SQL 扩展了类 Filesystem connector 对实时化用户场景和格式的支持，从而可以支持将流式数据从 Kafka 写入 Hive 等场景。

CDC 支持

CDC 格式是数据库中一种常用的模式，业务上典型的应用是通过工具（比如 Debezium 或 Canal）将 CDC 数据通过特定的格式从数据库中导出到 Kafka 中。在以前，业务上需要定义特殊的逻辑来解析 CDC 数据，并把它转换成一般的 Insert-only 数据，后续的处理逻辑需要考虑到这种特殊性，这种 work-around 的方式无疑给业务上带来了不必要的复杂性。如果 Flink SQL 引擎能原生支持 CDC 数据的输入，将 CDC 对接到 Flink SQL 的 Changelog Stream 概念上，将会大大降低用户业务的复杂度。

流计算的本质是就是不断更新、不断改变结果的计算。考虑一个简单的聚合 SQL，流计算中，每次计算产生的聚合值其实都是一个局部值，所以会产生 Changelog Stream。在以前想要把聚合的数据输出到 Kafka 中，如上图所示，几乎是不可能的，因为 Kafka 只能接收 Insert-only 的数据。Flink 之前主要是因为 Source&Sink 接口的限制，导致不能支持 CDC 数据的输入。

Flink SQL 1.11 经过了大量的接口重构，在新的 Source&Sink 接口上，支持了 CDC 数据的输入和输出，并且支持了 Debezium 与 Canal 格式（FLIP-105）。这一改动使动态 Table Source 不再只支持 append-only 的操作，而且可以导入外部的修改日志（插入事件）将它们翻译为对应的修改操作（插入、修改和删除）并将这些操作与操作的类型发送到后续的流中。

如上图所示，理论上，CDC 同步到 Kafka 的数据就是 Append 的一个流，只是在格式中含有 Changelog 的标识：

一种方式是把 Changlog 标识看做一个普通字段，这也是目前普遍的使用方式。
在 Flink 1.11 后，可以将它声明成 Changelog 的格式，Flink 内部机制支持 Interpret Changelog，可以原生识别出这个特殊的流，将其转换为 Flink 的 Changlog Stream，并按照 SQL 的语义处理；同理，Flink SQL 也具有输出 Change Stream 的能力（Flink 1.11 暂无内置实现），这就意味着，你可以将任意类型的 SQL 写入到 Kafka 中，只要有 Changelog 支持的 Format。
Flink 1.11 彻底的抛弃了推断 PK这个机制，不再从 Query 来推断 PK 了，而是完全依赖 Create table 语法。比如 Create 一个 jdbc_table，需要在定义中显式地写好 Primary Key（后面 NOT ENFORCED 的意思是不强校验，因为 Connector 也许没有具备 PK 的强校验的能力）。当指定了 PK，就相当于就告诉框架这个Jdbc Sink 会按照对应的 Key 来进行更新。如此，就跟 Query 完全没有关系了，这样的设计可以定义得非常清晰，如何更新完全按照设置的定义来。

Hive 流批一体

首先看传统的 Hive 数仓。一个典型的 Hive 数仓如下图所示。一般来说，ETL 使用调度工具来调度作业，比如作业每天调度一次或者每小时调度一次。这里的调度，其实也是一个叠加的延迟。调度产生 Table1，再产生 Table2，再调度产生 Table3，计算延时需要叠加起来。

问题是慢，延迟大，并且 Ad-hoc 分析延迟也比较大，因为前面的数据入库，或者前面的调度的 ETL 会有很大的延迟。Ad-hoc 分析再快返回，看到的也是历史数据。

所以现在流行构建实时数仓，从 Kafka 读计算写入 Kafka，最后再输出到 BI DB，BI DB 提供实时的数据服务，可以实时查询。Kafka 的 ETL 为实时作业，它的延时甚至可能达到毫秒级。实时数仓依赖 Queue，它的所有数据存储都是基于 Queue 或者实时数据库，这样实时性很好，延时低。但是：

第一，基于 Queue，一般来说就是行存加 Queue，存储效率其实不高。
第二，基于预计算，最终会落到 BI DB，已经是聚合好的数据了，没有历史数据。而且 Kafka 存的一般来说都是 15 天以内的数据，没有历史数据，意味着无法进行 Ad-hoc 分析。所有的分析全是预定义好的，必须要起对应的实时作业，且写到 DB 中，这样才可用。对比来说，Hive 数仓的好处在于它可以进行 Ad-hoc 分析，想要什么结果，就可以随时得到什么结果。

能否结合离线数仓和实时数仓两者的优势，然后构建一个 Lambda 的架构？

核心问题在于成本过高。无论是维护成本、计算成本还是存储成本等都很高。并且两边的数据还要保持一致性，离线数仓写完 Hive 数仓、SQL，然后实时数仓也要写完相应 SQL，将造成大量的重复开发。还可能存在团队上分为离线团队和实时团队，两个团队之间的沟通、迁移、对数据等将带来大量人力成本。如今，实时分析会越来越多，不断的发生迁移，导致重复开发的成本也越来越高。少部分重要的作业尚可接受，如果是大量的作业，维护成本其实是非常大的。

如何既享受 Ad-hoc 的好处，又能实现实时化的优势？一种思路是将 Hive 的离线数仓进行实时化，就算不能毫秒级的实时，准实时也好。所以，Flink 1.11 在 Hive 流批一体上做了一些探索和尝试，如下图所示。它能实时地按 Streaming 的方式来导出数据，写到 BI DB 中，并且这套系统也可以用分析计算框架来进行 Ad-hoc 的分析。这个图当中，最重要的就是 Flink Streaming 的导入。