Flink 实践教程：进阶10-自定义聚合函数（UDAF）

-- 建表语句，用于向 Source 提供数据
CREATE TABLE `udaf_input` (
  `id`       int(10) NOT NULL,
  `product`  varchar(50) DEFAULT '',
  `value`    int(10) DEFAULT NULL,
  `weight`   int(10) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
​
-- 插入数据
INSERT INTO `udaf_input` (`id`, `product`, `value`, `weight`) VALUES (1, 'oceanus-1', 2, 2);
INSERT INTO `udaf_input` (`id`, `product`, `value`, `weight`) VALUES (2, 'oceanus-1', 3, 3);
INSERT INTO `udaf_input` (`id`, `product`, `value`, `weight`) VALUES (3, 'oceanus-2', 5, 4);
INSERT INTO `udaf_input` (`id`, `product`, `value`, `weight`) VALUES (5, 'oceanus-2', 6, 5);
​
-- 建表语句，用于接收 Sink 端数据
CREATE TABLE `udaf_output` (
  `product`  varchar(50) NOT NULL DEFAULT '',
  `sum`      double(11,0) DEFAULT NULL,
  PRIMARY KEY (`product`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

package demos.UDAF;
​
public class WeightedAvgAccumulator{
    public long sum = 0;
    public int count = 0;
}

package demos.UDAF;
​
import org.apache.flink.table.functions.AggregateFunction;
​
public class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccumulator> {
​
    @Override
    public WeightedAvgAccumulator createAccumulator() {
        return new WeightedAvgAccumulator();
    }
​
    @Override
    public Long getValue(WeightedAvgAccumulator acc) {
        if (acc.count == 0) {
            return null;
        } else {
            return acc.sum / acc.count;
        }
    }
​
    public void accumulate(WeightedAvgAccumulator acc, Long iValue, Integer iWeight) {
        acc.sum += iValue * iWeight;
        acc.count += iWeight;
    }
​
    public void retract(WeightedAvgAccumulator acc, Long iValue, Integer iWeight) {
        acc.sum -= iValue * iWeight;
        acc.count -= iWeight;
    }
​
    public void merge(WeightedAvgAccumulator acc, Iterable<WeightedAvgAccumulator> it) {
        for (WeightedAvgAccumulator a : it) {
            acc.count += a.count;
            acc.sum += a.sum;
        }
    }
​
    public void resetAccumulator(WeightedAvgAccumulator acc) {
        acc.count = 0;
        acc.sum = 0L;
    }
}

mvn clean package

CREATE TEMPORARY SYSTEM FUNCTION WeightedAvg  AS 'demos.UDAF.WeightedAvg';

CREATE TABLE `mysql_cdc_source_table` (
  `id`        INT,
  `product`   VARCHAR,
  `value`     INT,
  `weight`    INT,
  PRIMARY KEY (`id`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
  'connector' = 'mysql-cdc',      -- 固定值 'mysql-cdc'
  'hostname' = 'xx.xx.xx.xx',      -- 数据库的 IP
  'port' = 'xxxx',                -- 数据库的访问端口
  'username' = 'root',            -- 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限）
  'password' = 'xxxxxxxxx',       -- 数据库访问的密码
  'database-name' = 'testdb',     -- 需要同步的数据库
  'table-name' = 'udaf_input'     -- 需要同步的数据表名
);

CREATE TABLE `jdbc_source_table` (
    `product`  VARCHAR,
    `sum`      DOUBLE,
    PRIMARY KEY(`product`) NOT ENFORCED
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://xx.xx.xx.xx:xxxx/testdb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',   -- 请替换为您的实际 MySQL 连接参数
    'table-name' = 'udaf_output',    -- 需要写入的数据表
    'username' = 'root',             -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'xxxxxxxxx',        -- 数据库访问的密码
    'sink.buffer-flush.max-rows' = '200',  -- 批量输出的条数
    'sink.buffer-flush.interval' = '2s'    -- 批量输出的间隔
);

INSERT INTO jdbc_source_table
SELECT
product,CAST(WeightedAvg(`value`,`weight`) AS DOUBLE) AS `sum`
FROM mysql_cdc_source_table GROUP BY `product`;