-- 建表语句，用于接收 Sink 端数据
CREATE TABLE `udf_output` (
  `id`        int(10) NOT NULL,
  `len_name`  int(10) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

开发 UDF

这里定义一个获取字符串字段长度的函数。如果传入一个字段，则获取这个字段的长度后返回；如果传入两个字段，则获取这两个字段的长度和后返回。

1. 代码编写

在本地IDE中创建 maven 项目，编写自定义函数UDF的代码。

// 类名：StringLengthUdf
package demos.UDF;

import org.apache.flink.table.functions.ScalarFunction;

public class StringLengthUdf extends ScalarFunction {

    public long eval(String a) {
        return a == null ? 0 : a.length();
    }

    public long eval(String b, String c) {
        return eval(b) + eval(c);
    }
}

2. 项目打包

使用 IDEA 自带打包工具 Build Artifacts 或者命令行进行打包。命令行打包命令：

mvn clean package

命令行打包后生成的 JAR 包可以在项目 target 目录下找到。

流计算 Oceanus 作业

上传依赖

在 Oceanus 控制台，点击左侧【依赖管理】，点击左上角【新建】新建依赖，上传本地 JAR 包。

创建 SQL 作业

在 Oceanus 控制台，点击左侧【作业管理】，点击左上角【新建】新建作业，作业类型选择 SQL 作业，点击【开发调试】进入作业编辑页面。单击【作业参数】，在【引用程序包】处选择刚才上传的 JAR 包。

1. 创建 Function

CREATE TEMPORARY SYSTEM FUNCTION StringLengthUdf AS 'demos.UDF.StringLengthUdf';

StringLengthUdf代表创建的函数名，demos.UDF.StringLengthUdf代表类路径。

2. 创建 Source

CREATE TABLE random_source ( 
  id INT, 
  name1 VARCHAR, 
  name2 VARCHAR 
  ) WITH ( 
  'connector' = 'datagen', 
  'rows-per-second'='1',          -- 每秒产生的数据条数
  'fields.id.kind'='sequence',    -- 无界的随机数
  'fields.id.start'='1',          -- 随机数的最小值
  'fields.id.end'='5',            -- 随机数的最大值
  'fields.name1.length'='10',     -- 随机字符串的长度
  'fields.name2.length'='10'      -- 随机字符串的长度
);

3. 创建 Sink

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

4. 编写业务 SQL

INSERT INTO jdbc_upsert_sink_table
SELECT
id,
CAST(StringLengthUdf(name1,name2) AS INT) AS `len_name`
FROM random_source;

运行作业

点击【发布草稿】->【运行版本】即可运行，可通过【日志】面板 TaskManager 或 Flink UI 查看运行信息。

总结

本文首先在本地开发 UDF 函数，将其打成 JAR 包后上传到 Oceanus 平台引用。接下来使用 Datagen 连接器产生虚拟数据，调用 UDF 函数进行不同字段的字符串长度的加和操作后存入 MySQL 中。

自定义标量函数（UDF）将0个、1个或多个标量值映射到一个新的标量值。
UDF 需要在 ScalarFunction 类中实现 eval 方法，且必须声明为 public 类型；自定义函数中 open 方法和 close 方法可选；可被重载，即在一个 UDF 中实现多个 eval 方法。

参考阅读

[1] 用户自定义函数 https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/table/functions/udfs

Flink系列

Flink 实践教程：进阶8-自定义标量函数（UDF）

流计算 Oceanus 简介

前置准备

创建流计算 Oceanus 集群

创建 MySQL 实例

创建 MySQL 表