Flink和StreamPark自定义UDF函数的使用

光头强鲁智深 發表於 2025-9-8 11:34:00

Flink和StreamPark自定义UDF函数的使用

本文分享自天翼云开发者社区《Flink和StreamPark自定义UDF函数的使用》，作者：王****帅
<h3 class="md-end-block md-heading md-focus">1、什么是函数</h3>
    在 SQL 中，我们可以把一些数据的转换操作包装起来，嵌入到 SQL 查询中统一调用，这就是“函数”（functions）。Flink 的 Table API 和 SQL 同样提供了函数的功能。两者在调用时略有不同：Table API 中的函数是通过数据对象的方法调用来实现的；而 SQL 则是直接引用函数名称，传入数据作为参数。例如，要把一个字符串 str 转换成全大写的形式，Table API 的写法是调用 str 这个 String对象的 upperCase()方法：
<pre class="language-bash"><code class="language-bash">str.upperCase();</code></pre>
    而 SQL 中的写法就是直接引用 UPPER()函数，将 str 作为参数传入：
<pre class="language-bash"><code class="language-bash">UPPER(str)</code></pre>
    由于 Table API 是内嵌在 Java 语言中的，很多方法需要在类中额外添加，因此扩展功能比较麻烦，目前支持的函数比较少；而且 Table API 也不如 SQL 的通用性强，所以一般情况下较少使用。下面我们主要介绍 Flink SQL 中函数的使用。Flink SQL 中的函数可以分为两类：一类是 SQL 中内置的系统函数，直接通过函数名调用就可以，能够实现一些常用的转换操作，比如之前我们用到的 COUNT()、CHAR_LENGTH()、UPPER()等等；而另一类函数则是用户自定义的函数（UDF），需要在表环境中注册才能使用。
<h3 class="md-end-block md-heading md-focus">2、什么是自定义UDF函数</h3>
    系统函数尽管庞大，也不可能涵盖所有的功能；如果有系统函数不支持的需求，我们就需要用自定义函数（User Defined Functions，UDF）来实现了。事实上，系统内置函数仍然在不断扩充，如果我们认为自己实现的自定义函数足够通用、应用非常广泛，也可以在项目跟踪工具 JIRA 上向 Flink 开发团队提出“议题”（issue），请求将新的函数添加到系统函数中。
<h4 class="md-end-block md-heading md-focus">2.1 编写自定义UDF函数</h4>
    自定义一个ScalarFunction，传入一个String类型的参数，输出这个参数的hashCode
<pre class="language-java"><code class="language-java">public class HashScalarFunction extends ScalarFunction {
public String eval(String str){
 return String.valueOf(str.hashCode());
}
}</code></pre>
<h4 class="md-end-block md-heading md-focus">2.2 在代码中以SQL方式使用UDF函数</h4>
<h5 class="md-end-block md-heading">2.2.1 读取mysql数据使用UDF函数转换并输出到控制台</h5>
<pre class="language-java"><code class="language-java">package cn.ctyun.demo.flinksql;

import cn.ctyun.demo.flinksql.udf.HashScalarFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
* @Date 2023/4/14 14:38
* @Description 读取mysql数据使用UDF函数转换并输出到控制台
*/
public class FlinkSqlUdfMysql2Print {

public static void main(String[] args) {

 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 // 1. 创建读取表，使用mysql进行
 String source_ddl = "CREATE TABLE UserSource (" +
 " id INT, " +
 " name VARCHAR, " +
 " phone VARCHAR, " +
 " sex INT " +
 ") WITH (" +
 " 'connector.type' = 'jdbc', " +
 " 'connector.url' = 'jdbc:mysql://*******:3306/flink_test_source?useSSL=false', " +
 " 'connector.table' = 'test_user_table', " +
 " 'connector.username' = 'root', " +
 " 'connector.password' = '******'" +
 ")";
 tableEnv.executeSql(source_ddl);
 // 3. 注册自定义标量函数
 tableEnv.createTemporarySystemFunction("MyHash", HashScalarFunction.class);
 // 4. 调用UDF查询转换
 Table resultTable = tableEnv.sqlQuery("select id, name, phone, sex, MyHash(name) as name_hash from UserSource");

 // 5. 输出到控制台
 tableEnv.executeSql("create table output (" +
 "id INT, " +
 "name STRING, " +
 "phone STRING, " +
 "sex INT, " +
 "name_hash STRING ) " +
 "WITH (" +
 "'connector' = 'print')");
 resultTable.executeInsert("output");
}
}</code></pre>
<h5 class="md-end-block md-heading">2.2.2 读取mysql数据使用UDF函数转换并输出到mysql</h5>
<pre class="language-java"><code class="language-java">package cn.ctyun.demo.flinksql;

import cn.ctyun.demo.flinksql.udf.HashScalarFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
* @Date 2023/4/14 14:50
* @Description 读取mysql数据使用UDF函数转换并输出到mysql
*/
public class FlinkSqlUdfMysql2Mysql {

public static void main(String[] args) {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 // 1. 创建读取表，使用mysql进行
 String source_ddl = "CREATE TABLE UserSource (" +
 " id INT, " +
 " name VARCHAR, " +
 " phone VARCHAR, " +
 " sex INT " +
 ") WITH (" +
 " 'connector.type' = 'jdbc', " +
 " 'connector.url' = 'jdbc:mysql://*******:3306/flink_test_source?useSSL=false', " +
 " 'connector.table' = 'test_user_table', " +
 " 'connector.username' = 'root', " +
 " 'connector.password' = '*******'" +
 ")";
 tableEnv.executeSql(source_ddl);

 //2. 创建写出表，使用mysql进行
 String sink_ddl = "CREATE TABLE UserSink (" +
 "id INT, " +
 "name STRING, " +
 "phone STRING, " +
 "sex INT, " +
 "name_hash STRING " +
 ") WITH (" +
 " 'connector.type' = 'jdbc', " +
 " 'connector.url' = 'jdbc:mysql://*******:3306/flink_test_sink?useSSL=false', " +
 " 'connector.table' = 'test_user_table_udf', " +
 " 'connector.username' = 'root', " +
 " 'connector.password' = '********'" +
 ")";
 tableEnv.executeSql(sink_ddl);

 // 3. 注册自定义标量函数
 tableEnv.createTemporarySystemFunction("MyHash", HashScalarFunction.class);

 // 4. 使用insert语句进行数据输出,在这里进行UDF查询转换
 String insertSql = "INSERT INTO UserSink select id, name, phone, sex, MyHash(name) as name_hash from UserSource";

 tableEnv.executeSql(insertSql);
}
}</code></pre>
<h4 class="md-end-block md-heading md-focus">2.3 在StreamPark中以SQL方式使用UDF函数</h4>
在StreamPark创建作业，导入作业依赖：
flink-connector-jdbc_2.12-1.14.3.jar
flink-demo-jar-job-1.0-SNAPSHOT.jar
mysql-connector-java-8.0.21.jar
FlinkSQL为：
<pre class="language-java"><code class="language-java">CREATE FUNCTION MyHash AS 'cn.ctyun.demo.flinksql.udf.HashScalarFunction';
CREATE TABLE UserSource (
id INT,
name VARCHAR,
phone VARCHAR,
sex INT
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://********:3306/flink_test_source?useSSL=false',
'connector.table' = 'test_user_table',
'connector.username' = 'root',
'connector.password' = '*********'
);
CREATE TABLE UserSink (
id INT,
name STRING,
phone STRING,
sex INT,
name_hash STRING
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://*******:3306/flink_test_sink?useSSL=false',
'connector.table' = 'test_user_table_udf',
'connector.username' = 'root',
'connector.password' = '**********'
);
INSERT INTO UserSink select id, name, phone, sex, MyHash(name) as name_hash from UserSource;</code></pre>
运行作业后mysql可正常插入数据 
来源：https://www.cnblogs.com/developer-tianyiyun/p/19079409

頁: [1]

琼殿技术论坛's Archiver

Flink和StreamPark自定义UDF函数的使用