技术专栏 flink关系型API: Table API 与SQL

  本篇文章主要介绍flink的关系型API,整个文章主要分为下面几个部分来介绍:

  flink已经拥有了强大的DataStream/DataSetAPI,满足流计算和批计算中的各种场景需求,但是关于以上几个问题,我们还需要提供一种关系型的API来实现Flink API层的流与批的统一,那么这就是flink的Table & SQL API。

  首先Table & SQL API是一种关系型API,用户可以像操作mysql数据库表一样的操作数据,而不需要写java代码完成Flink Function,更不需要手工的优化java代码调优。另外,SQL作为一个非程序员可操作的语言,学习成本很低,如果一个系统提供SQL支持,将很容易被用户接受。

  Table&SQLAPI是流处理和批处理统一的API层,如下图。flink在runtime层是统一的,因为flink将批任务看做流的一种特例来执行,然而在API层,flink为批和流提供了两套API(DataSet和DataStream)。所以Table&SQL API就统一了flink的API层,批数据上的查询会随着输入数据的结束而结束并生成DataSet,流数据的查询会一直运行并生成结果流。Table&SQL API做到了批与流上的查询具有同样的语法语义,因此不用改代码就能同时在批和流上执行。

  关于Table&SQL API,flink在0.9版本的时候,引进了TableAPI,支持Java和Scala两种语言,是一个类似于LINQ模式的API。用于对关系型数据进行处理。这系列 Table API的操作对象就是能够进行简单的关系型操作的结构化数据流。结构如下图。然而0.9版本的Table&SQL API有着很大的局限性,0.9版本Table API不能单独使用,必须嵌入到DataSet或者DataStream的程序中,对于批处理表的查询并不支持outer join、order by等操作。在流处理Table中只支持filters、union,不支持aggregations以及joins。并且,这个转化处理过程没有查询优化。整体来说0.9版本的Flink Table API还不是十分好用。

  在后续的版本中,1.1.0引入了 SQL,因此在1.1.0版本以后,flink 提供了两个语义的关系型 API:语言内嵌的 Table API(用于 Java 和Scala)以及标准 SQL。这两种 API 被设计用于在流和批的任务中处理数据在API层的统一,这意味着无论输入是批处理数据还是流数据,查询产生完全相同的结果。

  在1.20版本之后逐渐增加SQL的功能,并对Table API做了大量的Enhancement了。在1.2.0 版本中,flink 的关系 API在数据流中,支持关系操作包括投影、过滤和窗口聚合。

  优化逻辑计划并转换成flink的物理计划,flink的这部分实现统一封装在optimize方法里头。这部分涉及到多个阶段,每个阶段都是用Rule对逻辑计划进行优化和改进。声明定义于派生RelOptRule的一个类,然后再构造函数中要求传入RelOptRuleOperand对象,该对象需要传入这个Rule将要匹配的节点类型。如果这个自定义的Rule只用于LogicalTableScan节点,那么这个operand对象应该是operand(LogicalTableScan.class,any())。通过以上代码对逻辑计划进行了优化和转换,最后会将逻辑计划的每个节点转换成Flink Node,既可物理计划。

  (3)将一个外部的Catalog注册给TableEnvironment,访问外部系统的数据或文件。

  Table API是一个Scala和Java的集成查询序言。与SQL不同的是,Table API的查询不是一个指定的sql字符串,而是调用指定的API方法。TableAPI中的每一个方法输入都是一个Table,输出也是一个新的Table。通过table API来提交任务的话,也会经过calcite优化等阶段,基本流程和直接运行sql类似:

  Table API和SQL查询可以很容易的混合使用,因为它们的返回结果都是Table对象。一个基于Table API的查询可以基于一个SQL查询的结果。同样地,一个SQL查询可以被定义一个Table API注册TableEnvironment作为Table的查询结果。

  flink1.3以后,在flink sql上支持动态表查询,也就是说动态表是持续更新,并且能够像常规的静态表一样查询的表。但是,与批处理表查询终止后返回一个静态表作为结果不同的是,动态表中的查询会持续运行,并根据输入表的修改产生一个持续更新的表。因此,结果表也是动态的。在动态表中运行查询并产生一个新的动态表,这是因为流和动态表是可以相互转换的。

  流被转换为动态表,动态表使用一个持续查询进行查询,产生一个新的动态表。最后,结果表被转换成流。上面所说的只是逻辑模型,并不意味着实际执行的查询查询也是这个步骤。实际上,持续查询在内部被转换成传统的 DataStream 程序去执行。

  动态表上的 SQL 查询的第一步是在流中定义一个动态表。这意味着我们必须指定流中的记录如何修改现有的动态表。流携带的记录必须具有映射到表的关系模式。在流中定义动态表有两种模式:append模式和update模式。

  在append模式中,流中的每条记录是对动态表的插入修改。因此,流中的所有记录都append到动态表中,使得它的大小不断增长并且无限大。下图说明了append模式。append模式如下图。

  在update模式中,流中的记录可以作为动态表的插入、更新或者删除修改(append模式实际上是一种特殊的update模式)。当在流中通过update模式定义一个动态表时,我们可以在表中指定一个唯一的键属性。在这种情况下,更新和删除操作会带着键属性一起执行。更新模式如下图所示。

  一旦我们定义了动态表,我们可以在上面执行查询。由于动态表随着时间进行改变,我们必须定义查询动态表的意义。假定我们有一个特定时间的动态表的snapshot,这个snapshot可以作为一个标准的静态批处理表。我们将动态表 A 在点 t 的snapshot表示为 A[t],可以使用 SQL 查询来查询snapshot,该查询产生了一个标准的静态表作为结果,我们把在时间 t 对动态表 A 做的查询 q 的结果表示为 q(A[t])。如果我们反复在动态表的snapshot上计算查询结果,以获取进度时间点,我们将获得许多静态结果表,它们随着时间的推移而改变,并且有效的构成了一个动态表。我们在动态表的查询中定义如下语义:查询q 在动态表 A 上产生了一个动态表 R,它在每个时间点 t 等价于在 A[t] 上执行 q 的结果,即 R[t]=q(A[t])。该定义意味着在批处理表和流表上执行相同的查询 q 会产生相同的结果。

  在下面的例子中,我们给出了两个例子来说明动态表查询的语义。在下图中,我们看到左侧的动态输入表 A,定义成append模式。在时间t=8 时,A 由 6 行(标记成蓝色)组成。在时间 t=9 和 t=12 时,有一行追加到A(分别用绿色和橙色标记)。我们在表 A 上运行一个如图中间所示的简单查询,这个查询根据属性 k 分组,并统计每组的记录数。在右侧我们看到了 t=8(蓝色),t=9(绿色)和 t=12(橙色)时查询q 的结果。在每个时间点 t,结果表等价于在时间 t 时再动态表 A 上执行批查询。

  这个例子中的查询是一个简单的分组(但是没有窗口)聚合查询。因此,结果表的大小依赖于输入表的分组键的数量。此外,这个查询会持续更新之前产生的结果行,而不只是添加新行。

  第二个例子展示了一个类似的查询,但是有一个很重要的差异。除了对属性 k 分组以外,查询还将记录每 5 秒钟分组为一个滚动窗口,这意味着它每 5 秒钟计算一次 k 的总数。 我们使用 Calcite 的分组窗口函数来指定这个查询。在图的左侧,我们看到输入表 A ,以及它在append模式下随着时间而改变。在右侧,我们看到结果表,以及它随着时间演变。

  与第一个例子的结果不同的是,这个结果表随着时间而增长,例如每 5 秒钟计算出新的结果行。虽然非窗口查询更新结果表的行,但是窗口聚合查询只追加新行到结果表中。无论输入表什么时候更新,都不可能计算查询的完整结果。相反,查询编译成流应用,根据输入的变化持续更新它的结果。这意味着不是所有的有效 SQL 都支持,只有那些持续性的、递增的和高效计算的被支持。

  查询动态表生成的动态表,其相当于查询结果。根据查询和它的输入表,结果表会通过插入、更新和删除持续更改,就像普通的mysql数据表一样。它可能是一个不断被更新的单行表,或是一个只插入不更新的表。

  传统的mysql数据库在故障和复制的时候,通过日志重建表。比如 UNDO、REDO 和UNDO/REDO 日志。UNDO 日志记录被修改元素之前的值来回滚不完整的事务,REDO 日志记录元素修改的新值来重做已完成事务丢失的改变,UNDO/REDO 日志同时记录了被修改元素的旧值和新值来撤销未完成的事务,并重做已完成事务丢失的改变。基于这些日志,动态表可以转换成两类更改日志流:REDO 流和 REDO+UNDO 流。

  通过将表中的修改转换为流消息,动态表被转换为 redo+undo 流。插入修改生成一条新行的插入消息,删除修改生成一条旧行的删除消息,更新修改生成一条旧行的删除消息以及一条新行的插入消息。行为如下图所示。

  左侧显示了一个维护在append模式下的动态表,作为中间查询的输入。查询的结果转换为显示在底部的 redo+undo 流。输入表的第一条记录 (1,A) 作为结果表的一条新纪录,因此插入了一条消息 +(A,1) 到流中。第二条输入记录 k=‘A’(4,A) 导致了结果表中 (A,1) 记录的更新,从而产生了一条删除消息 -(A,1) 和一条插入消息 +(A,2)。所有的下游操作或数据汇总都需要能够正确处理这两种类型的消息。

  在两种情况下,动态表会转换成 redo 流:要么它只是一个append表(即只有插入修改),要么它有一个唯一的键属性。动态表上的每一个插入修改会产生一条新行的插入消息到 redo 流。由于 redo 流的限制,只有带有唯一键的表能够进行更新和删除修改。如果一个键从动态表中删除,要么是因为行被删除,要么是因为行的键属性值被修改了,所以一条带有被移除键的删除消息发送到 redo 流。更新修改生成带有更新的更新消息,比如新行。由于删除和更新修改根据唯一键来定义,下游操作需要能够根据键来访问之前的值。下图描述如何将上述相同查询的结果表转换为 redo 流。

  Redo 流的通常做法是将查询结果写到仅append的存储系统,比如滚动文件或者 Kafka topic ,或者是基于key访问的数据存储,比如 Cassandra、关系型 mysql。

  在1.2 版本中,flink 关系 API 的所有流操作,例如过滤和分组窗口聚合,只会产生新行,并且不能更新先前发布的结果。相比之下,动态表能够处理更新和删除修改。1.2 版本中的处理模型是动态表模型的一个子集, 通过附加模式将流转换为动态表,即一个无限增长的表。由于所有操作仅接受插入更改并在其结果表上生成插入更改(即,产生新行),因此所有在动态append表上已经支持的查询,将使用重做模型转换回 DataStreams,仅用于append表。

  最后,值得注意的是在开发代码中,我们无论是使用Table API还是SQL,优化和转换程序并不知道查询是通过 Table API 还是 SQL 来定义的。由于 Table API 和 SQL 在语义方面等同,只是在样式上有些区别而已。

  本篇文章整理了flink关系型API的相关知识,整体上来说,Flink关系型API还是很好用的,其原理与实现结构清晰,存在很多可借鉴的地方。

相关阅读