平顶山哪里做网站,带端口的服务器怎么做网站,3.建设营销型网站流程.,单页面网站怎么优化简介#xff1a; SQL 作为 Flink 中公认的核心模块之一#xff0c;对推动 Flink 流批一体功能的完善至关重要。在 1.11 中#xff0c;Flink SQL 也进行了大量的增强与完善#xff0c;开发大功能 10 余项#xff0c;不仅扩大了应用场景#xff0c;还简化了流程#xff0c…简介 SQL 作为 Flink 中公认的核心模块之一对推动 Flink 流批一体功能的完善至关重要。在 1.11 中Flink SQL 也进行了大量的增强与完善开发大功能 10 余项不仅扩大了应用场景还简化了流程上手操作更简单。
作者 | 李劲松Apache Flink Committer阿里巴巴技术专家
7 月 6 日Apache Flink 1.11 正式发布。从 3 月初进行功能规划到 7 月初正式发版1.11 用将近 4 个月的时间重点优化了 Flink 的易用性问题提升用户的生产使用体验。
SQL 作为 Flink 中公认的核心模块之一对推动 Flink 流批一体功能的完善至关重要。在 1.11 中Flink SQL 也进行了大量的增强与完善开发大功能 10 余项不仅扩大了应用场景还简化了流程上手操作更简单。
其中值得注意的改动包括
默认 Planner 已经切到 Blink planner 上。引入了对 CDCChange Data Capture变动数据捕获的支持用户仅用几句简单的 SQL 即可对接 Debezium 和 Canal 的数据源。离线数仓实时化用户可方便地使用 SQL 将流式数据从 Kafka 写入 Hive 等。
Flink SQL 演变
随着流计算的发展挑战不再仅限于数据量和计算量业务变得越来越复杂开发者可能是资深的大数据从业者、初学 Java 的爱好者或是不懂代码的数据分析者。如何提高开发者的效率降低流计算的门槛对推广实时计算非常重要。
SQL 是数据处理中使用最广泛的语言它允许用户简明扼要地展示其业务逻辑。Flink 作为流批一体的计算引擎致力于提供一套 SQL 支持全部应用场景Flink SQL 的实现也完全遵循 ANSI SQL 标准。之前用户可能需要编写上百行业务代码使用 SQL 后可能只需要几行 SQL 就可以轻松搞定。
Flink SQL 的发展大概经历了以下阶段
Flink 1.1.0第一次引入 SQL 模块并且提供 TableAPI当然这时候的功能还非常有限。Flink 1.3.0在 Streaming SQL 上支持了 Retractions显著提高了 Streaming SQL 的易用性使得 Flink SQL 支持了复杂的 Unbounded 聚合连接。Flink 1.5.0SQL 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 支持和高效的性能。
经过了多个版本的迭代支持SQL 模块在 Flink 中变得越来越重要Flink 的 SQL 用户也逐渐扩大。基于 SQL 模块的 Python 接口和机器学习接口也在快速发展。毫无疑问 SQL 模块作为最常用的 API 之一和生态的集成变得越来越重要。
SQL 1.11 重要变更
Flink SQL 在原有的基础上扩展了新场景的支持
Flink SQL 引入了对 CDCChange Data Capture变动数据捕获的支持它使 Flink 可以方便地通过像 Debezium 这类工具来翻译和消费数据库的变动日志。Flink SQL 扩展了类 Filesystem connector 对实时化用户场景和格式的支持从而可以支持将流式数据从 Kafka 写入 Hive 等场景。
除此之外Flink SQL 也从多个方面提高 SQL 的易用性系统性的解决了之前的 Bug、完善了用户 API。
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 之前主要是因为 SourceSink 接口的限制导致不能支持 CDC 数据的输入。Flink SQL 1.11 经过了大量的接口重构在新的 SourceSink 接口上支持了 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。
为了消费 CDC 数据用户需要在使用 SQL DDL 创建表时指指定“formatdebezium-json”或者“formatcanal-json”
CREATE TABLE my_table (...
) WITH (connector..., -- e.g. kafkaformatdebezium-json
);
Flink 1.11 的接口都已 Ready但是在实现上
只支持 Kafka 的 Debezium-json 和 Canal-json 读。欢迎大家扩展实现自己的 Format 和 Connector。
Source Sink 重构
Source Sink 重构的一个重要目的是支持上节所说的 Changelog但是除了 Changelog 以外它也解决了诸多之前的遗留问题。
新 Source Sink 使用标准姿势 详见官方文档
CREATE TABLE kafka_table (...
) WITH (connector kafka-0.10,topic test-topic,scan.startup.mode earliest-offset,properties.bootstrap.servers localhost:9092,format json,json.fail-on-missing-field false
);
Flink 1.11 为了向前兼容性依然保留了老 Source Sink使用 “connector.type” 的 Key即可 Fallback 到老 Source Sink 上。
■ Factory 发现机制
Flink 1.11 前用户可能经常遇到一个异常叫做 NoMatchingFactory 异常 指的是定义了一个 DDL在用的时候DDL 属性找不到对应的 TableFactory 实现可能的原因是
Classpath 下没有实现类Flink SQL 是通过 Java SPI 的机制来发现 Factory参数写错了。
但是报的异常让人非常疑惑根据异常的提示消息很难找到到底哪里的代码错了更难明确知道哪个 Key 写错了。
public interface Factory {String factoryIdentifier();……
}
所以在 Flink 1.11 中社区重构了 TableFactory 接口提出了一个新的 Factory 接口它有一个方法叫做 FactoryIdentifier。以后所有的 Factory 的 look up 都通过 identifier。这样的话就非常清晰明了找不到是因为 Classpath 下没 Factory 的类找得到那就可以定位到 Factory 的实现中进行确定性的校验。
■ 类型与数据结构
之前的 SourceSink 接口支持用户自定义数据结构即框架知道如何把自定义的数据结构转换为 Flink SQL 认识的内部数据结构如
public interface TableSourceT {TypeInformationT getReturnType();...
}
用户可以自定义泛型 T通过 getReturnType 来告诉框架怎么转换。
不过问题来了当 getReturnType 和 DDL 中声明的类型不一致时怎么办特别是两套类型系统的情况下如Runtime 的 TypeInformationSQL 层的 DataType。由于精度等问题可能导致经常出现类型不匹配的异常。
Flink 1.11 系统性地解决了这个问题。现在 Connector 开发者不能自定义数据结构只能使用 Flink SQL 内部的数据结构RowData。所以保证了默认 Type 与 DDL 的对应不用再返回类型让框架去确认了。
RowData 数据结构在 SQL 内部设计出来为了
抽象类接口在不同场景有适合的高性能实现。包含 RowKind契合流计算中的 CDC 数据格式。遵循 SQL 规范比如包含精度信息。对应 SQL 类型的可枚举的数据结构。■ Upsert 与 Primary Key
流计算的一个典型场景是把聚合的数据写入到 Upsert Sink 中比如 JDBC、HBase当遇到复杂的 SQL 时时常会出现 UpsertStreamTableSink 需要上游的 Query 有完整的 Primary Key 信息不然就直接抛异常。这个现象涉及到 Flink 的 UpsertStreamTableSink 机制。顾名思义它是一个更新的 Sink需要按 Key 来更新所以必须要有 Key 信息。
如何发现 Primary Key一个方法是让优化器从 Query 中推断如下图发现 Primary Key 的例子。
这种情况下在简单 Query 当中很好也满足语义也非常自然。但是如果是一个复杂的 Query比如聚合又 Join 再聚合那就只有报错了。不能期待优化器有多智能很多情况它都不能推断出 PK而且可能业务的 SQL 本身就不能推断出 PK所以导致了这样的异常。 怎么解决问题Flink 1.11 彻底的抛弃了这个机制不再从 Query 来推断 PK 了而是完全依赖 Create table 语法。比如 Create 一个 jdbc_table需要在定义中显式地写好 Primary Key后面 NOT ENFORCED 的意思是不强校验因为 Connector 也许没有具备 PK 的强校验的能力。当指定了 PK就相当于就告诉框架这个Jdbc Sink 会按照对应的 Key 来进行更新。如此就跟 Query 完全没有关系了这样的设计可以定义得非常清晰如何更新完全按照设置的定义来。
CREATE TABLE jdbc_table (id BIGINT,...PRIMARY KEY (id) NOT ENFORCED
)
Hive 流批一体
首先看传统的 Hive 数仓。一个典型的 Hive 数仓如下图所示。一般来说ETL 使用调度工具来调度作业比如作业每天调度一次或者每小时调度一次。这里的调度其实也是一个叠加的延迟。调度产生 Table1再产生 Table2再调度产生 Table3计算延时需要叠加起来。 问题是慢延迟大并且 Ad-hoc 分析延迟也比较大因为前面的数据入库或者前面的调度的 ETL 会有很大的延迟。Ad-hoc 分析再快返回看到的也是历史数据。
所以现在流行构建实时数仓从 Kafka 读计算写入 Kafka最后再输出到 BI DBBI 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 的导入。 ■ Streaming Sink
早期 Flink 版本在 DataStreaming 层已经有一个强大的 StreamingFileSink 将流数据写到文件系统。它是一个准实时的、Exactly-once 的系统能实现一条数据不多一条数据不少的 Sink。 具体原理是基于两阶段提交
第一阶段SnapshotPerTask关闭需要 Commit 的文件或者记录正在写的文件的 Offset。第二阶段NotifyCheckpointCompleteRename 需要 Commit 的文件。注意Rename 是一个原子且幂等的操作所以只要保证 Rename 的 At-least-once即可保证数据的 Exactly-once。
这样一个 File system 的 Writer 看似比较完美了。但是在 Hive 数仓中数据的可见性是依赖 Hive Metastore 的那在这个流程中谁来通知 Hive Metastore 呢 SQL 层在 StreamingFileSink扩展了 Partition 的 Committer。
相当于不仅要进行 File 的 Commit还要进行 Partition 的 Commit。如图所示FileWriter 对应之前的 StreamingFileSink它提供的是 Exactly-once 的 FileWriter。而后面再接了一个节点 PartitionCommitter。支持的 Commit Policy 有
内置支持 Add partition 到 Hive metastore支持写 SuccessFile 到文件系统当中并且也可以自定义 Committer比如可以 analysis partition、合并 partition 里面的小文件。
Committer 挂在 Writer 后 由 Commit Trigger 决定什么时机来 commit
默认的 commit 时机是有文件就立即 commit。因为所有 commit 都是可重入的所以这一点是可允许的。另外也支持通过 partition 时间和 Watermark 来共同决定的。比如小时分区如果现在时间到 11 点10 点的分区就可以 commit 了。Watermark 保证了作业当前的准确性。
■ Streaming Source
Hive 数仓中存在大量的 ETL 任务这些任务往往是通过调度工具来周期性的运行这样做主要有两个问题
实时性不强往往调度最小也是小时级。流程复杂组件多容易出现问题。
针对这些离线的 ETL 作业Flink 1.11 为此开发了实时化的 Hive 流读支持
Partition 表监控 Partition 的生成增量读取新的 Partition。非 Partition 表监控文件夹内新文件的生成增量读取新的文件。
甚至可以使用 10 分钟级别的分区策略使用 Flink 的 Hive streaming source 和 Hive streaming sink 可以大大提高 Hive 数仓的实时性到准实时分钟级在实时化的同时也支持针对 Table 全量的 Ad-hoc 查询提高灵活性。
SELECT * FROM hive_table /* OPTIONS(streaming-source.enable’true’, streaming-source.consume-start-offset2020-05-20) */;
另外除了 Scan 的读取方式Flink 1.11 也支持了 Temporal Join 的方式也就是以前常说的 Streaming Dim Join。
SELECTo.amout, o.currency, r.rate, o.amount * r.rate
FROMOrders AS oJOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS rON r.currency o.currency
目前支持的方式是 Cache All并且是不感知分区的比较适合小表的情况。
■ Hive Dialect
Flink SQL 遵循的是 ANSI-SQL 的标准而 Hive SQL 有它自己的 HQL 语法它们之间的语法、语义都有些许不同。
如何让 Hive 用户迁移到 Flink 生态中同时避免用户太大的学习成本为此 Flink SQL 1.11 提供了 Hive Dialect可以使得用户在 Flink 生态中使用 HQL 语言来计算。目前只支持 DDL后续版本会逐步攻坚 Qeuries。
■ Filesystem Connector
Hive Integration 提供了一个重量级的集成功能丰富但是环境比较复杂。如果只是想要一个轻量级的 Filesystem 读写呢
Flink table 在长久以来只支持一个 CSV 的 Filesystem Table并且还不支持 Partition行为上在某些方面也有些不符合大数据计算的直觉。
Flink 1.11 重构了整个 Filesystem connector 的实现
结合 Partition现在Filesystem connector 支持 SQL 中 Partition 的所有语义支持 Partition 的 DDL支持 Partition Pruning支持静态 / 动态 Partition 的插入支持 Overwrite 的插入。支持各种 FormatsCSVJSONAparch AVROApache ParquetApache ORC支持 Batch 的读写。支持 Streaming sink也支持 Partition commit支持写 Success 文件。
用几句简单的 SQL不用搭建 Hive 集成环境即可
启动一个流作业写入 Filesystem 中然后在 Hive 端即可查询到 Filesystem 上的数据相比之前 Datastream 的作业简单 SQL 即可搞定离线数据的入库。通过 Filesystem Connector 来查询 Hive 数仓中的数据功能没有 Hive 集成那么全但是定义简单。
Table 易用性
■ DDL Hints 和 Like 在 Flink 1.10 以后Hive MetaStore 逐渐成为 Flink streaming SQL 中 Table 相关的 Meta 信息的存储。比如可以通过 Hive Catalog 保存 Kafka Tables。这样可以在启动的时候直接使用 Tables。
通过 DDL 这种方式把 SQL 提交到 Cluster就可以写入 Kafka或者写入 MySQL、 DFS。使用 Hive Catalog 后是不是说只用写一次 DDL之后的流计算作业都是直接使用 Kafka 的 Table 呢
不完全是因为还是有一些缺陷。比如一个典型的 Kafka Table 有一些 Execution 相关的参数。因为 kafka 一般来说都是存 15 天以内的数据需要指定每次消费的时间偏移时间偏移是在不断变化的。每次提交作业使用 Kafka Table 的参数是不一样的。而这些参数又存储在 Catalog 里面这种情况下只能创建另外一张表所以字段和参数要重写一遍非常繁琐。
Flink 1.11社区就开发了 Table Hints它在 1.11 中目前只专注一个功能即 Dynamic Table Options。用起来很简单在 SQL 中 Select From 时在 Table 后面写 Table Hints 的方式来指定其动态 Options 在不同的使用场景指定不同的动态参数。Flink 1.11引入了 Like 语法。LIKE 是标准的 SQL 定义。相当于 Clone 一张表出来复用它的 schema。LIKE 支持多种 Constraints。可以选择继承也可以选择完全覆盖。
Table Hints
SELECT id, name FROM kafka_table1 /* OPTIONS(scan.startup.modeearliest-offset) */;
LIKE
CREATE TABLE kafka_table2 WITH ( scan.startup.modeearliest-offset) LIKE kafka_table1;
这两个手段在对接 Hive Catalog 的基础上是非常好的补充能够尽可能的避免在每次作业的时候都写一大堆 Schema。
■ 内置 Connectors
Flink SQL 1.11 引入了新的三个内置 Connectors主要是为了大家更方便的进行调试、测试以及进行压测和线上的观察
Datagen Source一个无中生有产生数据的 Source可以定义生成的策略比如 Sequence比如 Random 的生成。方便线下进行功能性的测试也可以拿来性能测试。Print Sink直接在 Task 节点 Runtime 的打印出数据比如线上作业某个 Sink 少数据了不知道是上游发来数据有问题还是 Sink 逻辑有问题这时可以额外接一个 Print Sink排查上游数据到底有没有问题。Blackhole Sink默默把数据给吃掉方便功能性的调试。
这三个 Connectors 的目的是为了在调试、测试中排除 Connectors 的影响一般来说Connectors 在流计算中是不可控的存在很多问题把 Connectors 糅杂在一起变得比较复杂难以排查。
SQL-API
■ TableEnvironment
TableEnvironment 作为 SQL 层的编程入口无疑是非常重要的之前的 API 主要是
Table sqlQuery从一段 Select 的 Query 中返回 Table 接口把用户的 SQL 翻译成 Flink 的 Table。void sqlUpdate本质上是执行一段 DDL/DML。但是行为上当是 DDL 时直接执行当是 DML 时默默 Cache 到 TableEnvironment等到后续的 execute 调用才会真正的执行。execute真正的执行提交作业到集群。
TableEnvironment 默默的 Cache 执行计划而且多个 API 感觉上会很混乱所以1.11 社区重构了编程接口目的是想要提供一个干净、并且不易出 bug 的清晰接口。
单 SQL 执行TableResult executeSql(String sql)多 SQL 执行StatementSet createStatementSet()TableResult支持 collect、print、getJobClient
现在 executeSql 就是一个大一统的接口不管是什么 SQL是 Query 还是 DDL 还是 DML直接丢给它都可以很方便地使用起来。
并且和 Datastream 也有了很清晰的界限
调用过 toDataStream一定要使用 StreamExecutionEnvironment.execute没调用过 toDataStream一定要使用 TableEnvironment.executeSql
■ SQL-Client
SQL-Client 在 1.11 对齐了很多 Flink 内部本来就支持的 DDL除此之外值得注意的是社区还开发了 Tableau 的结果展示模式展示更自然一些直接在命令行展示结果而不是切换页面 总结和展望
上述解读主要侧重在用户接口方面社区已经有比较丰富的文档大家可以去官网查看这些功能的详细文档以便更深入的了解和使用。
Flink SQL 1.11 在 CDC 方面开了个头内部机制和 API 上打下了夯实的基础未来会内置更多的 CDC 支持直接对接数据库 Binlog支持更多的 Flink SQL 语法。后续版本也会从底层提供更多的流批一体支持给 SQL 层带来更多的流批一体的可能性。 原文链接 本文为阿里云原创内容未经允许不得转载。
