分析函数
分析函数会保持状态来做分析工作。在流式处理规则中,分析函数会首先被执行,这样它们就不会受到 WHERE 子句的影响而必不更新状态。
分析函数完整使用格式如下,其中 over 子句为可选子句。
AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>] [WHEN <Expression>])
分析函数的计算是在当前查询输入的所有输入事件上进行的,可以选择限制分析函数只考虑符合 PARTITION BY 子句的事件。 分析函数可以使用 PARTITION BY 子句,语法如下:
AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>])
分析函数可以使用 WHEN 条件判断子句,根据是否满足条件来确定当前事件是否为有效事件。 当为有效事件时,根据分析函数语意计算结果并更新状态。当为无效事件时,忽略事件值,复用保存的状态值。
AnalyticFuncName(<arguments>...) OVER ([WHEN <Expression>])
LAG
lag(expr, [offset], [default value], [ignore null])
返回表达式前一个值在偏移 offset 处的结果,如果没有找到,则返回默认值,如果没有指定默认值则返回 nil。
示例1:获取之前温度值的函数
lag(temperature)
示例2:获取相同设备之前温度值的函数
lag(temperature) OVER (PARTITION BY deviceId)
示例3:ts为时间戳,获取设备状态 statusCode1 和 statusCode2 不相等持续时间
select lag(Status) as Status, ts - lag(ts, 1, ts, true) OVER (WHEN had_changed(true, statusCode)) as duration from demo
LATEST
latest(expr, [default value])
返回表达式最新的非空值。如果没有找到,则返回默认值。否则,返回 nil 。
CHANGED_COL
changed_col(true, col)
返回列的相比上次执行后的变化值。若未变化则返回 null 。
HAD_CHANGED
had_changed(true, expr1, expr2, ...)
返回是否上次运行后列的值有变化。 其参数可以为 * 以方便地监测所有列。
监控变化的函数
Changed_col 函数
该函数为普通的标量函数,因此可在任意的子句,包括 SELECT 和 WHERE 中使用。
语法
CHANGED_COL(<ignoreNull>, <expr>)
参数
ignoreNull: 判断变化时是否忽略 null 值。若为 true,则收到 null 值或未收到值不会触发变化。
expr: 用来监控变化状态和输出变化值的表达式。
返回值
返回变化后的值或者 null (未变化)。与所有标量函数相同,该函数默认返回的列名未函数的名字 changed_col 。可使用 as alias
赋别名。
Changed_cols 函数
该函数返回多个列的结果,因此只能在 SELECT 子句中使用。
语法
CHANGED_COLS (<prefix>, <ignoreNull>, <expr> [,...,<exprN>])
参数
prefix: 返回的列名的前缀。默认情况下,返回的变化列名与原列名相同,例如 CHANGED_COLS("", true, col1)
返回 col1
。如果设置了前缀参数,则返回的列名将加上前缀以区别于普通的列,例如 CHANGED_COLS("changed_", true, col1)
将返回 changed_col1
。
ignoreNull: 判断变化时是否忽略 null 值。若为 true,则收到 null 值或未收到值不会触发变化。
expr: 用来监控变化状态和输出变化值的表达式。可以为任何可在 SELECT 子句中使用的表达式。若表达式为 *
则会返回所有列的变化。
返回值
返回所有与上一次运行的值有变化的表达式的新值。如果在普通规则中使用,则与上次事件触发时的值比较。如果在窗口规则中使用,则与上次窗口输出的值比较。
首次运行时,返回所有表达式的值,因为没有前一次的运行,所有表达式都判定为有变化。
在接下来的运行中,如果选择的所有表达式都没有值变化,则返回空值。
注意事项
多列函数仅可在 select 子句中使用。其选出的值不能用于 WHERE 或其他子句中。若需要根据变化值做过滤,则应使用 CHANGED_COL 函数,或者将 CHANGED_COLS 的规则作为规则流水线的前置规则。
函数返回的列命别名仅能通过 prefix 参数做全局的设置。若需要给每个列设置单独的别名,则需要使用 CHANGED_COL 函数。
Had_changed 函数
该函数为向量函数,支持不定长度参数。
HAD_CHANGED (<ignoreNull>, <expr> [,...,<exprN>])
参数
ignoreNull: 判断变化时是否忽略 null 值。若为 true,则收到 null 值或未收到值不会触发变化。
expr: 用来监控变化状态和输出变化值的表达式。可以为任何可在 SELECT 子句中使用的表达式。若表达式为 *
则监测所有列的变化。
返回值
返回一个 bool 值,表示上次运行后的变化状态。多参数版本与用或连接使用单个参数的版本相同,即 HAD_CHANGED(expr1) OR HAD_CHANGED(expr2) ... OR HAD_CHANGED(exprN) 。若需要监测别的关系,可单独使用此函数。例如,监测是否所有值都有变化,可使用 HAD_CHANGED(expr1) AND HAD_CHANGED(expr2) ... AND HAD_CHANGED(exprN) 。
范例
创建流 demo,并给与如下输入。
{
"ts": 1,
"temperature": 23,
"humidity": 88
}
{
"ts": 2,
"temperature": 23,
"humidity": 88
}
{
"ts": 3,
"temperature": 23,
"humidity": 88
}
{
"ts": 4,
"temperature": 25,
"humidity": 88
}
{
"ts": 5,
"temperature": 25,
"humidity": 90
}
{
"ts": 6,
"temperature": 25,
"humidity": 91
}
{
"ts": 7,
"temperature": 25,
"humidity": 91
}
{
"ts": 8,
"temperature": 25,
"humidity": 91
}
获取 temperature 变化值的规则:
SQL: SELECT CHANGED_COLS("", true, temperature) FROM demo
___________________________________________________
{"temperature":23}
{"temperature":25}
获取 temperature 或 humidity 的变化值并添加名称前缀的规则:
SQL: SELECT CHANGED_COLS("c_", true, temperature, humidity) FROM demo
_________________________________________________________
{"c_temperature":23,"c_humidity":88}
{"c_temperature":25}
{"c_humidity":90}
{"c_humidity":91}
获取所有列的变化值并且不忽略 null 值的规则:
SQL: SELECT CHANGED_COLS("c_", false, *) FROM demo
_________________________________________________________
{"c_ts":1, "c_temperature":23, "c_humidity":88}
{"c_ts":2}
{"c_ts":3}
{"c_ts":4, "c_temperature":25}
{"c_ts":5, "c_humidity":90}
{"c_ts":6, "c_humidity":91}
{"c_ts":7}
{"c_ts":8}
获取窗口中平均值变化的规则:
SQL: SELECT CHANGED_COLS("t", true, avg(temperature)) FROM demo GROUP BY CountWindow(2)
_________________________________________________________________
{"tavg":23}
{"tavg":24}
{"tavg":25}
当 temperature 或者 humidity 变化时获取数据:
SQL: SELECT ts, temperature, humidity FROM demo
WHERE HAD_CHANGED(true, temperature, humidity) = true
_________________________________________________________
{"ts":1,temperature":23,"humidity":88}
{"ts":4,temperature":25,"humidity":88}
{"ts":5,temperature":25,"humidity":90}
{"ts":6,temperature":25,"humidity":91}
当 temperature 变化且 humidity 未变化时获取数据:
SQL: SELECT ts, temperature, humidity FROM demo
WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
_________________________________________________________
{"ts":4,temperature":25,"humidity":88}
获取 temperature 和 humidity 的变化值并赋自定义名:
SQL: SELECT CHANGED_COL(true, temperature) AS myTemp, CHANGED_COL(true, humidity) AS myHum FROM demo
_________________________________________________________
{"myTemp":23,"myHum":88}
{"myTemp":25}
{"myHum":90}
{"myHum":91}
当 temperature 值变化后大于 24 时获取数据:
SQL: SELECT ts, temperature, humidity FROM demo
WHERE CHANGED_COL(true, temperature) > 24
_________________________________________________________
{"ts":4,temperature":25,"humidity":88}
ACC 函数
ACC 函数全称为 accumulate function,该函数将会根据所得的参数进行累计计算,累计范围为该规则的整个生命周期。
对于接下来的 acc 函数,我们将用以下数据进行模拟输入输出:
a
依次输入 3 条数据,分别为 1,2,3。
ACC_SUM
acc_sum(expr)
acc_sum 函数对表达式结果进行累计加和,返回累计加和结果。
示例1:使用 acc_sum 进行累计加和
acc_sum(a)
结果为分别为: 1 3 6
ACC_MAX
acc_max(expr)
acc_max 函数对表达式结果进行累计比较取较大值,返回累计比较取较大值的结果。
示例1:使用 acc_max 进行累计比较取较大值
acc_max(a)
结果为分别为: 1 2 3
ACC_MIN
acc_min(expr)
acc_min 函数对表达式结果进行累计比较取较小值,返回累计比较取较小值的结果。
示例1:使用 acc_min 进行累计比较取较小值
acc_min(a)
结果为分别为: 1 1 1
ACC_COUNT
acc_count(expr)
acc_count 函数对表达式结果进行累计个数统计,返回累计个数值。
示例1:使用 acc_count 进行累计个数统计
acc_count(a)
结果为分别为: 1 2 3
ACC_AVG
acc_avg(expr)
acc_avg 函数对表达式结果进行累计平均值统计,返回累计平均值。
示例1:使用 acc_count 进行累计平均值统计
acc_avg(a)
结果为分别为: 1 1.5 2