西宁高端网站建设,江苏建筑模板厂家,软件开发和网站建设哪个好,公司内部网站系统简介
TDengine 中的流计算#xff0c;功能相当于简化版的 FLINK #xff0c; 具有实时计算#xff0c;计算结果可以输出到超级表中存储#xff0c;同时也可用于窗口预计算#xff0c;加快查询速度。
创建流式计算
CREATE STREAM [IF NOT EXISTS] stream_name [stream_o…简介
TDengine 中的流计算功能相当于简化版的 FLINK 具有实时计算计算结果可以输出到超级表中存储同时也可用于窗口预计算加快查询速度。
创建流式计算
CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name[(field1_name, field2_name [PRIMARY KEY], ...)] [TAGS (create_definition [, create_definition] ...)] SUBTABLE(expression) AS subquery
stream_options: {TRIGGER [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time | FORCE_WINDOW_CLOSE]WATERMARK timeIGNORE EXPIRED [0|1]DELETE_MARK timeFILL_HISTORY [0|1]IGNORE UPDATE [0|1]
}
其中 subquery 是 select 普通查询语法的子集。
subquery: SELECT select_listfrom_clause[WHERE condition][PARTITION BY tag_list]window_clause支持会话窗口、状态窗口、滑动窗口、事件窗口和计数窗口。其中状态窗口、事件窗口和计数窗口搭配超级表时必须与 partition by tbname 一起使用。对于数据源表是复合主键的流不支持状态窗口、事件窗口、计数窗口的计算。
stb_name 是保存计算结果的超级表的表名如果该超级表不存在会自动创建如果已存在则检查列的 schema 信息。详见 写入已存在的超级表。
TAGS 子句定义了流计算中创建TAG的规则可以为每个 partition 对应的子表生成自定义的TAG值详见 自定义 TAG
create_definition:col_name column_definition
column_definition:type_name [COMMENT string_value]subtable 子句定义了流式计算中创建的子表的命名规则详见 流式计算的 partition。
window_clause: {SESSION(ts_col, tol_val)| STATE_WINDOW(col)| INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)]| EVENT_WINDOW START WITH start_trigger_condition END WITH end_trigger_condition| COUNT_WINDOW(count_val[, sliding_val])
}其中 SESSION 是会话窗口tol_val 是时间间隔的最大范围。在 tol_val 时间间隔范围内的数据都属于同一个窗口如果连续的两条数据的时间超过 tol_val则自动开启下一个窗口。该窗口的 _wend 等于最后一条数据的时间加上 tol_val。 STATE_WINDOW 是状态窗口col 用来标识状态量相同的状态量数值则归属于同一个状态窗口col 数值改变后则当前窗口结束自动开启下一个窗口。 INTERVAL 是时间窗口又可分为滑动时间窗口和翻转时间窗口。INTERVAL 子句用于指定窗口相等时间周期SLIDING 字句用于指定窗口向前滑动的时间。当 interval_val 与 sliding_val 相等的时候时间窗口即为翻转时间窗口否则为滑动时间窗口注意sliding_val 必须小于等于 interval_val。 EVENT_WINDOW 是事件窗口根据开始条件和结束条件来划定窗口。当 start_trigger_condition 满足时则窗口开始直到 end_trigger_condition 满足时窗口关闭。start_trigger_condition 和 end_trigger_condition 可以是任意 TDengine 支持的条件表达式且可以包含不同的列。 COUNT_WINDOW 是计数窗口按固定的数据行数来划分窗口。count_val 是常量是正整数必须大于等于 2小于 2147483648。count_val 表示每个 COUNT_WINDOW 包含的最大数据行数总数据行数不能整除 count_val 时最后一个窗口的行数会小于 count_val。sliding_val 是常量表示窗口滑动的数量类似于 INTERVAL 的 SLIDING。
窗口的定义与时序数据特色查询中的定义完全相同详见 TDengine 特色查询
例如如下语句创建流式计算。第一个流计算自动创建名为 avg_vol 的超级表以一分钟为时间窗口、30 秒为前向增量统计这些电表的平均电压并将来自 meters 表的数据的计算结果写入 avg_vol 表不同 partition 的数据会分别创建子表并写入不同子表。
第二个流计算自动创建名为 streamt0 的超级表将数据按时间戳的顺序以 voltage 0 作为窗口的开始条件voltage 9 作为窗口的结束条件划分窗口做聚合运算并将来自 meters 表的数据的计算结果写入 streamt0 表不同 partition 的数据会分别创建子表并写入不同子表。
第三个流计算自动创建名为 streamt1 的超级表将数据按时间戳的顺序以 10 条数据为一组划分窗口做聚合运算并将来自 meters 表的数据的计算结果写入 streamt1 表不同 partition 的数据会分别创建子表并写入不同子表。
CREATE STREAM avg_vol_s INTO avg_vol AS
SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(1m) SLIDING(30s);CREATE STREAM streams0 INTO streamt0 AS
SELECT _wstart, count(*), avg(voltage) from meters PARTITION BY tbname EVENT_WINDOW START WITH voltage 0 END WITH voltage 9;CREATE STREAM streams1 IGNORE EXPIRED 1 WATERMARK 100s INTO streamt1 AS
SELECT _wstart, count(*), avg(voltage) from meters PARTITION BY tbname COUNT_WINDOW(10);流式计算的 partition
可以使用 PARTITION BY TBNAMEtag普通列或者表达式对一个流进行多分区的计算每个分区的时间线与时间窗口是独立的会各自聚合并写入到目的表中的不同子表。
不带 PARTITION BY 子句时所有的数据将写入到一张子表。
在创建流时不使用 SUBTABLE 子句时流式计算创建的超级表有唯一的 tag 列 groupId每个 partition 会被分配唯一 groupId。与 schemaless 写入一致我们通过 MD5 计算子表名并自动创建它。
若创建流的语句中包含 SUBTABLE 子句用户可以为每个 partition 对应的子表生成自定义的表名例如
CREATE STREAM avg_vol_s INTO avg_vol SUBTABLE(CONCAT(new-, tname)) AS SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname tname INTERVAL(1m);PARTITION 子句中为 tbname 定义了一个别名 tname, 在 PARTITION 子句中的别名可以用于 SUBTABLE 子句中的表达式计算在上述示例中流新创建的子表将以前缀 ‘new-’ 连接原表名作为表名从 v3.2.3.0 开始为了避免 SUBTABLE 中的表达式无法区分各个子表即误将多个相同时间线写入一个子表在指定的子表名后面加上 _stableName_groupId。
注意子表名的长度若超过 TDengine 的限制将被截断。若要生成的子表名已经存在于另一超级表由于 TDengine 的子表名是唯一的因此对应新子表的创建以及数据的写入将会失败。
流式计算读取历史数据
正常情况下流式计算不会处理创建前已经写入源表中的数据若要处理已经写入的数据可以在创建流时设置 fill_history 1 选项这样创建的流式计算会自动处理创建前、创建中、创建后写入的数据。流计算处理历史数据的最大窗口数是 2000 万超过限制会报错。例如
create stream if not exists s1 fill_history 1 into st1 as select count(*) from t1 interval(10s)结合 fill_history 1 选项可以实现只处理特定历史时间范围的数据例如只处理某历史时刻2020 年 1 月 30 日之后的数据
create stream if not exists s1 fill_history 1 into st1 as select count(*) from t1 where ts 2020-01-30 interval(10s)再如仅处理某时间段内的数据结束时间可以是未来时间
create stream if not exists s1 fill_history 1 into st1 as select count(*) from t1 where ts 2020-01-30 and ts 2023-01-01 interval(10s)如果该流任务已经彻底过期并且您不再想让它检测或处理数据您可以手动删除它被计算出的数据仍会被保留。
删除流式计算
DROP STREAM [IF EXISTS] stream_name;仅删除流式计算任务由流式计算写入的数据不会被删除。
展示流式计算
SHOW STREAMS;若要展示更详细的信息可以使用
SELECT * from information_schema.ins_streams;流式计算的触发模式
在创建流时可以通过 TRIGGER 指令指定流式计算的触发模式。
对于非窗口计算流式计算的触发是实时的对于窗口计算目前提供 4 种触发模式默认为 WINDOW_CLOSE。 AT_ONCE写入立即触发 WINDOW_CLOSE窗口关闭时触发窗口关闭由事件时间决定可配合 watermark 使用 MAX_DELAY time若窗口关闭则触发计算。若窗口未关闭且未关闭时长超过 max delay 指定的时间则触发计算。 FORCE_WINDOW_CLOSE以操作系统当前时间为准只计算当前关闭窗口的结果并推送出去。窗口只会在被关闭的时刻计算一次后续不会再重复计算。该模式当前只支持 INTERVAL 窗口不支持滑动FILL_HISTORY 必须为 0IGNORE EXPIRED 必须为 1IGNORE UPDATE 必须为 1FILL 只支持 PREV、NULL、NONE、VALUE。
由于窗口关闭是由事件时间决定的如事件流中断、或持续延迟则事件时间无法更新可能导致无法得到最新的计算结果。
因此流式计算提供了以事件时间结合处理时间计算的 MAX_DELAY 触发模式。MAX_DELAY 最小时间是 5s如果低于 5s创建流计算时会报错。
MAX_DELAY 模式在窗口关闭时会立即触发计算。此外当数据写入后计算触发的时间超过 max delay 指定的时间则立即触发计算
流式计算的窗口关闭
流式计算以事件时间插入记录中的时间戳主键为基准计算窗口关闭而非以 TDengine 服务器的时间以事件时间为基准可以避免客户端与服务器时间不一致带来的问题能够解决乱序数据写入等等问题。流式计算还提供了 watermark 来定义容忍的乱序程度。
在创建流时可以在 stream_option 中指定 watermark它定义了数据乱序的容忍上界。
流式计算通过 watermark 来度量对乱序数据的容忍程度watermark 默认为 0。
T 最新事件时间 - watermark
每次写入的数据都会以上述公式更新窗口关闭时间并将窗口结束时间 T 的所有打开的窗口关闭若触发模式为 WINDOW_CLOSE 或 MAX_DELAY则推送窗口聚合结果。 图中纵轴表示不同时刻对于不同时刻我们画出其对应的 TDengine 收到的数据即为横轴。
横轴上的数据点表示已经收到的数据其中蓝色的点表示事件时间(即数据中的时间戳主键)最后的数据该数据点减去定义的 watermark 时间得到乱序容忍的上界 T。
所有结束时间小于 T 的窗口都将被关闭图中以灰色方框标记。
T2 时刻乱序数据黄色的点到达 TDengine由于有 watermark 的存在这些数据进入的窗口并未被关闭因此可以被正确处理。
T3 时刻最新事件到达T 向后推移超过了第二个窗口关闭的时间该窗口被关闭乱序数据被正确处理。
在 window_close 或 max_delay 模式下窗口关闭直接影响推送结果。在 at_once 模式下窗口关闭只与内存占用有关。
流式计算对于过期数据的处理策略
对于已关闭的窗口再次落入该窗口中的数据被标记为过期数据.
TDengine 对于过期数据提供两种处理方式由 IGNORE EXPIRED 选项指定 增量计算即 IGNORE EXPIRED 0。 直接丢弃即 IGNORE EXPIRED 1默认配置忽略过期数据
无论在哪种模式下watermark 都应该被妥善设置来得到正确结果直接丢弃模式或避免频繁触发重算带来的性能开销重新计算模式。
流式计算对于修改数据的处理策略
TDengine 对于修改数据提供两种处理方式由 IGNORE UPDATE 选项指定 检查数据是否被修改即 IGNORE UPDATE 0如果数据被修改则重新计算对应窗口。 不检查数据是否被修改全部按增量数据计算即 IGNORE UPDATE 1默认配置。
写入已存在的超级表
[field1_name, ...]在本页文档顶部的 [field1_name, …] 是用来指定 stb_name 的列与 subquery 输出结果的对应关系的。如果 stb_name 的列与 subquery 输出结果的位置、数量全部匹配则不需要显示指定对应关系。如果 stb_name 的列与 subquery 输出结果的数据类型不匹配会把 subquery 输出结果的类型转换成对应的 stb_name 的列的类型。创建流计算时不能指定 stb_name 的列和 TAG 的数据类型否则会报错。
对于已经存在的超级表检查列的schema信息
检查列的 schema 信息是否匹配对于不匹配的则自动进行类型转换当前只有数据长度大于 4096byte 时才报错其余场景都能进行类型转换。检查列的个数是否相同如果不同需要显示的指定超级表与 subquery 的列的对应关系否则报错如果相同可以指定对应关系也可以不指定不指定则按位置顺序对应。
自定义 TAG
用户可以为每个 partition 对应的子表生成自定义的 TAG 值。
CREATE STREAM streams2 trigger at_once INTO st1 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st partition by concat(tag-, tbname) as cc interval(10s));PARTITION 子句中为 concat(“tag-”, tbname) 定义了一个别名 cc对应超级表 st1 的自定义 TAG 的名字。在上述示例中流新创建的子表的 TAG 将以前缀 ‘new-’ 连接原表名作为 TAG 的值。
会对TAG信息进行如下检查
检查 tag 的 schema 信息是否匹配对于不匹配的则自动进行数据类型转换当前只有数据长度大于 4096byte 时才报错其余场景都能进行类型转换。检查 tag 的个数是否相同如果不同需要显示的指定超级表与 subquery 的 tag 的对应关系否则报错如果相同可以指定对应关系也可以不指定不指定则按位置顺序对应。
清理中间状态
DELETE_MARK timeDELETE_MARK 用于删除缓存的窗口状态也就是删除流计算的中间结果。如果不设置默认值是 10 年 T 最新事件时间 - DELETE_MARK
流式计算支持的函数
所有的 单行函数 均可用于流计算。以下 19 个聚合/选择函数 不能 应用在创建流计算的 SQL 语句。此外的其他类型的函数均可用于流计算。
leastsquarespercentiletopbottomelapsedinterpderivativeiratetwahistogramdiffstatecountstatedurationcsummavgsampletailuniquemode
暂停、恢复流计算
1.流计算暂停计算任务 PAUSE STREAM [IF EXISTS] stream_name; 没有指定 IF EXISTS如果该 stream 不存在则报错如果存在则暂停流计算。指定了 IF EXISTS如果该 stream 不存在则返回成功如果存在则暂停流计算。
2.流计算恢复计算任务 RESUME STREAM [IF EXISTS] [IGNORE UNTREATED] stream_name; 没有指定 IF EXISTS如果该 stream 不存在则报错如果存在则恢复流计算指定了 IF EXISTS如果 stream 不存在则返回成功如果存在则恢复流计算。如果指定 IGNORE UNTREATED则恢复流计算时忽略流计算暂停期间写入的数据。
状态数据备份与同步
流计算的中间结果成为计算的状态数据需要在流计算整个生命周期中进行持久化保存。为了确保流计算中间状态能够在集群环境下在不同的节点间可靠地同步和迁移从 v3.3.2.1 开始需要在运行环境中部署 rsync 软件还需要增加以下的步骤
在配置文件中配置 snode 的地址IP 端口和状态数据备份目录该目录系 snode 所在的物理节点的目录。然后创建 snode。 完成上述两个步骤以后才能创建流。 如果没有创建 snode 并正确配置 snode 的地址流计算过程中将无法生成检查点checkpoint并可能导致后续的计算结果产生错误。 snodeAddress 127.0.0.1:873 checkpointBackupDir /home/user/stream/backup/checkpoint/ 创建 snode 的方式
使用以下命令创建 snodestream nodesnode 是流计算中有状态的计算节点可用于部署聚合任务同时负责备份不同的流计算任务生成的检查点数据。
CREATE SNODE ON DNODE [id]其中的 id 是集群中的 dnode 的序号。请注意选择的dnode流计算的中间状态将自动在其上进行备份。 从 3.3.4.0 版本开始在多副本环境中创建流会进行 snode 的存在性检查要求首先创建 snode。如果 snode 不存在无法创建流。
访问官网
更多内容欢迎访问 TDengine 官网
