AI 助理
备案控制台
文档
输入文档关键字查找
首页 实时计算 Flink版 操作指南 运维管理 作业状态管理 Flink状态兼容性 SQL修改及兼容性 变更Query

Group Aggregate

更新时间：
产品详情
我的收藏

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

可兼容的变更

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

    • 对于新增统计指标，属于部分兼容变更，从当前作业启动时开始累计。

    • 对于删除统计指标，属于完全兼容变更。删除的统计指标对应的状态数据会被丢弃。

    • 对于既有新增又有删除统计指标，则属于部分兼容变更。新增的统计指标从当前作业启动时开始累计，删除的统计指标对应的状态数据会被丢弃。

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

    说明

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- 新增统计指标：count(c)，属于部分兼容变更。
-- sum(b)、max(c) 的计算结果不受影响，count(c) 的值在作业启动时从 0 开始累计
SELECT a, SUM(b), MAX(c), COUNT(c) FROM MyTable GROUP BY a;

-- 删除统计指标：sum(b)，属于完全兼容变更。
-- max(c)的计算结果不受影响。
SELECT a, MAX(c) FROM MyTable GROUP BY a;

-- 修改统计指标：max(c) -> min(c)，属于部分兼容变更。
-- sum(b) 的计算结果不受影响。max(c) 被认为删除，其对应的状态数据会被丢弃。
-- min(c) 被认为是新增指标，其值在作业启动时开始计算。
SELECT a, SUM(b), MIN(c) FROM MyTable GROUP BY a;

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- 调整统计指标位置：sum(b)、max(c)，属于完全兼容变更。
-- sum(b)、max(c)的计算结果不受影响。
SELECT a, MAX(c), SUM(b) FROM MyTable GROUP BY a;

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- 修改统计指标：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;

  • 多个包含DISTINCT聚合函数和非DISTINCT聚合函数更换顺序，但是DISTINCT聚合函数相对顺序一致，属于完全兼容变更。

    -- 原始SQL。
INSERT INTO MySink
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;

-- SUM(DISTINCT b)仍然在COUNT(DISTINCT b)前，相对顺序没变，属于完全兼容变更。
INSERT INTO MySink
SELECT a, SUM(DISTINCT b), COUNT(DISTINCT b), MAX(b)  FROM MyTable GROUP BY a;

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

  • 在Retract流后删除Retract聚合函数，属于完全兼容变更。

    -- 原始SQL。
SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- 删除Retract聚合函数，属于完全兼容变更。
SELECT c/2, AVG(avg_a) AS avg_avg_a FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

不兼容的变更

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- 新增统计维度：d，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a, d;

-- 删除统计维度：a，当前修改属于不兼容变更。
SELECT SUM(b), MAX(c) FROM MyTable;

-- 修改统计维度：a -> d，当前修改属于不兼容变更。
SELECT d, SUM(b), MIN(c) FROM MyTable GROUP BY d;

-- 修改了统计维度：a -> a + 1，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c) FROM (
  SELECT a + 1 AS a, b, c FROM MyTable 
) GROUP BY a;

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- 新增distinct统计指标：count(distinct b)，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- 删除distinct统计指标：sum(distinct b)，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- 修改distinct统计指标：sum(distinct b) -> avg(distinct b)，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c), AVG(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- 修改distinct统计指标：count(distinct c) -> count(distinct avg(c))，当前修改属于不兼容变更。
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM (
     SELECT a, b, AVG(c) AS c from MyTable GROUP BY a, b
) GROUP BY a;

  • 级联聚合时，新增统计指标，该修改属于不兼容变更。（级联聚合会发生消息撤回，新增统计指标结果不可预期。）

    -- 原始SQL。
SELECT a/2, AVG(b), MIN(c) FROM (
    SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
) GROUP BY a/2;

-- 新增统计指标：count(c)，当前修改属于不兼容变更。
SELECT a/2, AVG(b), MIN(c), COUNT(c) FROM (
    SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
) GROUP BY a/2;

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

    -- 原始SQL。
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- 删除所有统计指标：sum(b), max(c)，当前修改属于不兼容变更。
SELECT a FROM MyTable GROUP BY a;

  • 多个包含DISTINCT聚合函数和非DISTINCT聚合函数更换顺序，但是DISTINCT聚合函数相对顺序变化，属于不兼容变更。

    -- 原始SQL。
INSERT INTO MySink
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;

--SUM(DISTINCT b)被交换到了COUNT(DISTINCT b)后，相对顺序发生变化，属于完全兼容变更。
INSERT INTO MySink
SELECT COUNT(DISTINCT b), a, MAX(b), SUM(DISTINCT b) FROM MyTable GROUP BY a;

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

    -- 原始SQL。
INSERT INTO MySink
SELECT a, b FROM MyTable GROUP BY a,b;

-- 加入新的聚合函数，属于不兼容变更。
INSERT INTO MySink
SELECT a, b, SUM(b) FROM MyTable GROUP BY a,b;

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

    -- 原始SQL。
INSERT INTO MySink
SELECT a, SUM(b), MAX(b), MAX(c) FROM MyTable GROUP BY a;

--修改统计指标和计算逻辑，属于不兼容变更。
INSERT INTO MySink 
SELECT a, MAX(c) FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable) GROUP BY a;

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

    -- 原始SQL。
INSERT INTO MySink
SELECT a, b, MAX(c) FROM MyTable GROUP BY a,b;

-- 修改统计指标，属于不兼容变更。
INSERT INTO MySink
SELECT a, b, MIN(c) FROM MyTable GROUP BY a,b;

  • 在Retract流后新增或变更Retract聚合函数，属于不兼容变更。

    -- 原始SQL。
SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- 新增Retract聚合函数，属于不兼容变更。
SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) min_max_b, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- 变更Retract聚合函数，属于不兼容变更。
SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

未知兼容变更

存在Python自定义聚合函数（UDAF）时，任何变更都属于未知兼容变更。

-- 作业变更前后，存在例如python自定义weighter_avg的函数，属于未知兼容。
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b), weighted_avg(a, b) 
FROM MyTable GROUP BY a, b;
上一篇：SQL修改及兼容性下一篇：变更Source
文档推荐