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

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

// 类名：StringLengthUdfpackage 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);  }}

mvn clean package

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

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'      -- 随机字符串的长度);

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'    -- 批量输出的间隔);

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