GroupWindowAggregate变更兼容性 - 实时计算Flink版

本文为您介绍Group Window Aggregate变更的可兼容性和不可兼容性详情。

可兼容的变更

新增、删除、修改非Distinct的统计指标（Aggregate Function）。

对于新增统计指标，属于部分兼容，从当前作业启动时开始累计。
对于删除统计指标，属于完全兼容修改。删除的统计指标对应的状态数据会被丢弃。
对于既有新增又有删除统计指标，则属于部分兼容修改。新增的统计指标从当前作业启动时开始累计，删除的统计指标对应的状态数据会被丢弃。
对于修改统计指标，被视为删除和新增两个操作，属于部分兼容。新增的统计指标从当前作业启动时开始累计，删除的统计指标对应的状态数据会被丢弃。

说明

对于未进行任何变更的统计指标，复用状态数据后计算的结果和基于历史数据运行的结果是一致的。

-- 原始SQL。
select a, sum(b), max(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

-- 新增统计指标：count(c)，属于部分兼容修改。
-- sum(b)、max(c)的计算结果不受影响，count(c)的值在作业启动时从0开始累计。
select a, sum(b), max(c), count(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

-- 删除统计指标：sum(b)，属于完全兼容修改。
-- max(c) 的计算结果不受影响
select a, max(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

-- 修改统计指标：max(c) -> min(c)，属于部分兼容修改。
-- sum(b)的计算结果不受影响。max(c)被认为删除，其对应的状态数据会被丢弃。
-- min(c)被认为是新增指标，其值在作业启动时开始计算。
select a, sum(b), min(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

调整非Distinct的统计指标位置，该修改属于完全兼容变更。

-- 原始SQL。
select a, sum(b), max(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

-- 调整统计指标位置：sum(b)、max(c)，属于完全兼容修改。
-- sum(b)、max(c) 的计算结果不受影响。
select a, max(c), sum(b)
from MyTable
group by a, tumble(ts, interval '1' minute);

当非Distinct的统计指标涉及的字段有计算逻辑变化时，该统计指标被认为进行了修改操作，该修改属于部分兼容变更。

-- 原始SQL。
select a, sum(b), max(c)
from MyTable
group by a, tumble(ts, interval '1' minute);

-- 修改统计指标：max(c) -> max(substring(c, 1, 5))，属于部分兼容修改。
-- sum(b)的计算结果不受影响。max(c)被认为删除，其对应的状态数据会被丢弃。
-- max(substring(c, 1, 5))被认为新增指标，其值在作业启动时开始计算。
select a, sum(b), max(c)
from (select a, b, substring(c, 1, 5) as c from MyTable)
group by a, tumble(ts, interval '1' minute);

新增、删除window属性字段，该修改属于完全兼容变更。

-- 原始SQL。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 新增window end属性，该修改属于完全兼容变更。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start,
  tumble_end(ts, interval '1' minute) as window_end
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 删除window start属性，该修改属于完全兼容修改。
select a,
  sum(b),
  max(c)
from MyTable
  group by a, tumble(ts, interval '1' minute);

修改Group Key顺序，除窗口函数外其他Group Key相对顺序一致，属于完全兼容变更。

-- 原始SQL。
select a, sum(distinct b), max(distinct c), count(c)
  from MyTable 
group by a, b, tumble(rowtime, interval '15' minute);

-- 修改Group Key顺序，且相对顺序一致，属于完全兼容变更。
select a, sum(distinct b), max(distinct c), count(c)
  from MyTable 
group by a, tumble(rowtime, interval '15' minute), b;

作业变更前后都没有统计指标，属于完全兼容变更。

不兼容的变更

修改window相关属性（window的类型，window的大小，时间相关属性），该修改属于不兼容变更。

-- 原始SQL。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 修改window类型：tumble -> hop，当前修改属于不兼容修改。
select a,
  sum(b),
  max(c),
  hop_start(ts, interval '1' minute, interval '2' minute) as window_start
from MyTable
  group by a, hop(ts, interval '1' minute, interval '2' minute);

-- 修改window 大小：'1' minute -> '2' minute，当前修改属于不兼容修改。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '2' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '2' minute);

-- 修改时间相关属性：ts -> proctime，当前修改属于不兼容修改。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(proctime, interval '1' minute);

新增、删除、修改统计维度（group key）或者统计维度涉及字段的计算逻辑发生变化，该修改属于不兼容变更。

-- 原始SQL。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 新增统计维度：d，当前修改属于不兼容修改。
select a,
  sum(b),
  max(c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, d, tumble(ts, interval '1' minute);

-- 其他示例可参考Group Aggregate修改group key示例。

新增、删除、修改Distinct的统计指标（distinct aggregate function）或者Distinct统计指标涉及字段的计算逻辑发生变化，该修改属于不兼容变更。

-- 原始SQL。
select a,
  sum(b),
  max(c),
  count(distinct c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 新增Distinct统计指标：count(distinct b)，当前修改属于不兼容修改。
select a,
  sum(b),
  count(distinct b),
  max(c),
  count(distinct c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);
-- 其他示例可参考Group Aggregate修改distinct统计指标示例。

删除了所有统计指标，该修改属于不兼容变更。（所有统计指标的状态数据都被丢弃，没有状态复用。）

-- 原始SQL。
select a,
  sum(b),
  count(distinct b),
  max(c),
  count(distinct c),
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

-- 删除所有统计指标：sum(b), max(c)，当前修改属于不兼容修改。
select a,
  tumble_start(ts, interval '1' minute) as window_start
from MyTable
  group by a, tumble(ts, interval '1' minute);

当添加、删除、修改了earlyFire或lateFire时，该修改属于不兼容变更。

修改Group Key顺序，除窗口函数外其他Group Key相对顺序发生变化，属于不兼容变更。

-- 原始SQL。
select a, sum(distinct b), max(distinct c), count(c)
  from MyTable 
group by a, b, tumble(rowtime, interval '15' minute);

-- 修改Group Key顺序，普通Group Key间顺序变化，属于不兼容变更。
select a, sum(distinct b), max(distinct c), count(c)
  from MyTable 
group by b, a, tumble(rowtime, interval '15' minute);

变更前作业没有任何一个统计指标，而新作业加入新的统计指标，属于不兼容变更。

-- 原始SQL。
select a, b, c
  from MyTable 
group by a, b, c, tumble(rowtime, interval '15' minute);

--加入新的统计指标，属于不兼容变更。
select a, b, c, count(c)
  from MyTable 
group by a, b, c, tumble(rowtime, interval '15' minute);

变更后最终仅有一个统计指标，并且修改了计算逻辑，属于不兼容变更。

-- 原始SQL。
select a, sum(b), max(b), max(c) 
  from MyTable 
group by b, a, tumble(ts, interval '15' minute);

-- 新作业仅有一个统计指标，并且修改计算逻辑，属于不兼容变更。
select a, max(c) 
  from (select a, b, c + 1 as c, ts from MyTable) 
group by b, a, tumble(ts, interval '15' minute);

变更前后任何一个统计指标都不同，属于不兼容变更。

-- 原始SQL。
select a, sum(b), max(b), max(c) from MyTable 
group by a, tumble(ts, interval '1' second);

--统计指标完全不同，属于不兼容变更。
select a, min(b), avg(b) from MyTable 
group by a, tumble(ts, interval '1' second);

变更前后只要存在参数table.exec.emit.early-fire.enabled为true或者table.exec.emit.late-fire.enabled为true，属于不兼容变更。

-- 原始SQL。
select a, max(c) 
from (select a, b, c + 1 as c, ts from MyTable) 
group by b, a, TUMBLE(ts, interval '15' minute);

-- 添加参数，属于不兼容变更。
set 'table.exec.emit.early-fire.enabled' = 'true';
set 'table.exec.emit.early-fire.delay' = '500ms';
-- 或者
set 'table.exec.emit.late-fire.enabled' = 'true';
set 'able.exec.emit.late-fire.delay' = '1s';
set 'table.exec.emit.allow-lateness' = '5s';

select a, max(c) 
from (select a, b, c + 1 as c, ts from MyTable) 
group by b, a, TUMBLE(ts, interval '15' minute);

未知兼容变更

变更前后只要存在Python自定义聚合函数（UDAF），属于未知兼容。

-- 作业变更前后，存在例如python自定义weighter_avg的函数，属于未知兼容。
SELECT COUNT(DISTINCT b), a, SUM(DISTINCT b),weighted_avg(a, b) 
FROM MyTable GROUP BY a, c;