Flink Table/SQL API 规划 —— Dynamic Table

动态表的概念是社区很早就提出的但并没有全部实现下文中所有介绍都是基于已有规划和proposal给出的,可能与之后实现存在出入仅供参考

概念

动态表直观上看是一个类似于数据库中的Materialized View概念。动态表随着时间改变;类似静态的batch table一样可以用标准SQL进行查询然后一个新的动态表;可以和流无损地互相转换(对偶的)。对现有的API最大的改进关键在表的内容随着时间改变,而现在的状态只是append。当前的streaming table可以认为是一种动态表,append模式的动态表。

流到 Dynamic Table

流被转换成Table时决定选择哪种模式是依据表的schema是否定义primary key。

Append模式:

如果表的schema没有包括key的定义那转换成表时采用append模式。把流中每条新来的record当做新的row append到表中。一旦数据加到表中就不能再被更新和删除(指当前表中,不考虑转换成新表)。

Replace模式:

相对应,如果定义了key,那么对于流中的每条记录如果key不在表中就insert否则就update。

Dynamic Table 到 流

表到流的操作是把表的所有change以changelog stream的方式发送到下游。这一步也有两种模式。

Retraction模式:

traction模式中对于Dynamic Table的insert和delete的change分别产生insert或delete event。如果是update的change会产生两种change event,对于之前发送出去的同样key的record会产生delete event,对于当前的record是产生insert event。如下图所示:

Update模式:

update模式依赖Dynamic Table定义了key。所有的change event是一个kv对。key对应表的key在当前record中的值;对于insert和change value对应新的record。对于delete value是空表示该可以已经被删除。如下图所示:

example

表的内容随着时间改变意味着对表的query结果也是随着时间改变的。我们定义:

  • A[t]: 时间t时的表A
  • q(A[t]):时间t时对表A执行query q

举个例子来理解动态表的概念:

query的限制

由于流是无限的,相对应 Dynamic Table 也是无界的。当查询无限的表的时候我们需要保证query的定时是良好的,有意义可行的。

1.在实践中Flink将查询转换成持续的流式应用,执行的query仅针对当前的逻辑时间,所以不支持对于任意时间点的查询(A[t])。
2.最直观的原则是query可能的状态和计算必须是有界的,所以可以支持可增量计算的查询:

  • 不断更新当前结果的查询:查询可以产生insert,update和delete更改。查询可以表示为 Q(t+1) = q'(Q(t), c(T, t, t+1)),其中Q(t)是query q的前一次查询结果,c(T, t, t_+1) 是表T从t+1到t的变化, q'是q的增量版本。
  • 产生append-only的表,可以从输入表的尾端直接计算出新数据。查询可以表示为 Q(t+1) = q''(c(T, t-x, t+1)) ∪ Q(t),q''是不需要时间t时q的结果增量版本query q。c(T, t-x, t+1)是表T尾部的x+1个数据,x取决于语义。例如最后一小时的window aggregation至少需要最后一小时的数据作为状态。其他能支持的查询类型还有:单独在每一行上操作的SELECT WHERE;rowtime上的GROUP BY子句(比如基于时间的window aggregate);ORDER BY rowtime的OVER windows(row-windows);ORDER BY rowtime。
    3.当输入表足够小时,对表的每条数据进行访问。比如对两个大小固定的流表(比如key的个数固定)进行join。

中间状态有界

如上文所说的,某些增量查询需要保留一些数据(部分输入数据或者中间结果)作为状态。为了保证query不会失败,保证查询所需要的空间是有界的不随着时间无限增长很重要。主要有两个原因使得状态增长:
1. 不受时间谓词约束的中间计算状态的增长(比如 聚合key的膨胀)
2. 时间有界但是需要迟到的数据(比如 window 的聚合)

虽然第二种情况可有通过下文提到的"Last Result Offset"参数解决,但是第一种情况需要优化器检测。我们应该拒绝不受时间限制的中间状态增长的查询。优化器应该提供如何修复查询且要求有适当的时间谓词。比如下面这个查询:

SELECT user, page, COUNT(page) AS pCnt
FROM pageviews
GROUP BY user, page

随着用户数和页面数的增长,中间状态会数据随着时间推移而增长。对于存储空间的要求可以通过添加时间谓词来限制:

SELECT user, page, COUNT(page) AS pCnt
FROM pageviews
WHERE rowtime BETWEEN now() - INTERVAL '1' HOUR AND now() // only last hour
GROUP BY user, page

因为不是所有属性都是不断增长的, 因此可以告诉优化器domain的size, 就可以推断中间状态不会随着时间推移而增长,然后接受没有时间谓词的查询。

val sensorT: Table = sensors
  .toTable('id, 'loc, 'stime, 'temp)
  .attributeDomain('loc, Domain.constant) // domain of 'loc is not growing
env.registerTable("sensors", sensorT)

SELECT loc, AVG(temp) AS avgTemp
FROM sensors
GROUP BY loc

结果的计算和细化时序

一些关系运算符必须等数据到达才能计算最终结果。例如:在10:30关闭的窗口至少要等到10:30才能计算出最终的结果。Flink的logical clock(即 决定何时才是10:30)取决于使用event time 还是 processing time。在processing time的情况下,logical time是每个机器的wallclock;在event time的情况下,logical clock time是由源头提供的watermark决定的。由于数据的乱序和延迟当在event time模式下时等待一段时间来减小计算结果不完整性。另一方面某些情况下希望得到不断改进的早期结果。因此对于结果被计算、改进或者做出最终结果时有不同的要求、

下图描绘了不同的配置参数如何用于控制早期结果和细化计算结果的。

  • "First Result Offset" 指第一个早期结果被计算的结果的时间。时间是相对于第一次可以计算完整结果的时间(比如相对于window的结束时间10:30)。如果设置的是-10分钟,对于结束时间是10:30的window那么第一个被发出去的结果是在逻辑时间10:20计算的。这个参数的默认值是0,即在window结束的时候才计算结果。
  • "Complete Result Offset" 指完整的结果被计算的时间。时间是相对于第一次可以计算完整的时间。如果设置的是+5分钟,对于结束时间是10:30的window那么产生完整结果的时间是10:35。这个参数可以减轻延迟数据造成的影响。默认是0,即在window结束的时候计算的结果就是完整结果。
  • "Update Rate" 指计算完整结果之前一次次更新结果的时间间隔(可以是时间和次数)。如果设为5分钟,窗口大小是30分钟的tumbling window,开始时间是10:300,"First Result Offset"是-15分钟, "Complete Result Offset"是2分钟,那么将在10:20, 10:25, 10:30更新结果,10:15禅城寄一个结果,10:32产生完整结果。
  • "Last Updates Switch" 指完整结果发出后对于延迟的数据是否计算延迟更新,直到计算状态被清除。
  • "Last Result Offset" 指可计算的最后一个结果的时间。这是内部状态被清除的时间,清除状态后再到达的数据将被丢弃。Last Result Offset 意味着计算的结果是近似值,不能保证精确。

原文链接

时间: 2024-10-02 20:16:38

Flink Table/SQL API 规划 —— Dynamic Table的相关文章

CREATE TABLE – SQL Command

command CREATE TABLE – SQL CommandCreates a table having the specified fields. CREATE TABLE | DBF TableName1 [NAME LongTableName] [FREE] (FieldName1 FieldType [(nFieldWidth [, nPrecision])] [NULL | NOT NULL] [CHECK lExpression1 [ERROR cMessageText1]]

C# 我用SQL在几张table里面抓了一些数据放到EXCLE模板里,通过判断栏位的值加底色

问题描述 C# 我用SQL在几张table里面抓了一些数据放到EXCLE模板里,通过判断栏位的值加底色 我用SQL在几张table里面抓了一些数据放到EXCLE里,现在表里面有CBRS_REQUIRED这个列,通过判断CBRS_REQUIRED这个列里面的值是否是'yes',如果是'yes'的话则给yes的这一行底色改成红色.我现在只能给excel所有的行都加上红色底色,希望大家帮我加上这个判断.谢谢,下面是代码,底色我写在最后. using System; using System.Text;

UCenter info: MySQL Query Error SQL:SELECT value FROM [Table]vars WHERE noteexists_Mysql

大家先看下数据库权限问题,然后再进行如下操作. SQL:SELECT value FROM [Table]vars WHERE name='noteexists2′ UCenter info: MySQL Query Error SQL:SELECT value FROM [Table]vars WHERE name='noteexists2′ Error:SELECT command denied to user '数据库'@'IP地址' for table 'pre_ucenter_vars

sql入门教程:sql 创建表create table 与数据库[create databse]

sql入门教程:sql 创建表create table 与数据库[create databse] 在创建数据库的声明 在CREATE DATABASE语句是用来建立一个数据库. 创建数据库的SQL语法 CREATE DATABASE database_name 创建数据库的例子现在,我们要建立一个数据库,所谓的" my_db " . 我们使用下面的CREATE DATABASE语句:CREATE DATABASE my_db   下面来看看sql 创建表create table 在创建

使用Informix Dynamic Server调度器和SQL API

新的数据库管理特性简介 IBM IDS 11提供了一组令人期待的新特性,这组新特性形成了RDBMS自管理的基础.这些工具使DBA可以收集服务器信息,并使用SQL作为远程管理工具调整数据库服务器.此外,调度器还可以对 SQL API 命令的执行进行调度,这很像 UNIX cron 任务.IBM IDS 本身已是声誉卓著的免管理数据库,如今又降低了管理与问题诊断的成本,其价值更上一个台阶. IBM IDS 11实现的管理方面的增强有: 一个新的系统数据库(sysadmin) 数据库调度器 管理性SQ

mysql不预读数据库信息(use dbname 更快,不会卡)—Reading table information for completion of table and column names

原文:http://blog.163.com/xychenbaihu@yeah/blog/static/13222965520113152619796/ mysql> use dbname Reading table information for completion of table and column names You can turn off this feature to get a quicker startup with -A 让后就卡在这里.     上面卡住的原因::   

向table的tr加个table,加不上??

问题描述 data.bznr里是一个table;xEditingArea是个div的id;var table='<table id="main" width="100%" style="table-layout:fixed;height:100%;"'+ 'cellSpacing="0"cellPadding="0"><tbody><tr vAlign="top&qu

一样的table?不一样的table(可编辑状态table)_jquery

新的一天开始了,生活还要继续,今天要和大家分享的是不一样的table,普通的table就用于显示数据,今天要分享的table不仅仅能显示数据,还可以对数据进行编辑,当鼠标点击数据时相应的数据格就变成可编辑的状态,废话不多说,进入今天的主题,首先完成HTML页面: 复制代码 代码如下: <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/html4/strict.dtd"

新手问一个在C#里读取 sql 的view 和 table之间的transform出现的行数问题

问题描述 我在做一个transform的项目,以前用的是一个view,现在一步步转换成使用table,在C#里面原有的计算方法全部都会有所改变,问题发生在view里面使用view.AddNew(),加新建一行数据的时候数据库里面是没有写入的.(行号一直都是0)直到使用Factory.Save();之后就会在SQL新建这一行数据.(行号变成1)然而在原本unittest单元测试的时候,往往不会使用Factory.save();来写入数据但是从view转换成table后,一旦使用了自定义的metho