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

本文分享自天翼云开发者社区《Flink和StreamPark自定义UDF函数的使用》，作者：王****帅

1、什么是函数

    在 SQL 中，我们可以把一些数据的转换操作包装起来，嵌入到 SQL 查询中统一调用，这就是“函数”（functions）。Flink 的 Table API 和 SQL 同样提供了函数的功能。两者在调用时略有不同：Table API 中的函数是通过数据对象的方法调用来实现的；而 SQL 则是直接引用函数名称，传入数据作为参数。例如，要把一个字符串 str 转换成全大写的形式，Table API 的写法是调用 str 这个 String对象的 upperCase()方法：

str.upperCase();

    而 SQL 中的写法就是直接引用 UPPER()函数，将 str 作为参数传入：

UPPER(str)

    由于 Table API 是内嵌在 Java 语言中的，很多方法需要在类中额外添加，因此扩展功能比较麻烦，目前支持的函数比较少；而且 Table API 也不如 SQL 的通用性强，所以一般情况下较少使用。下面我们主要介绍 Flink SQL 中函数的使用。Flink SQL 中的函数可以分为两类：一类是 SQL 中内置的系统函数，直接通过函数名调用就可以，能够实现一些常用的转换操作，比如之前我们用到的 COUNT()、CHAR_LENGTH()、UPPER()等等；而另一类函数则是用户自定义的函数（UDF），需要在表环境中注册才能使用。

2、什么是自定义UDF函数

    系统函数尽管庞大，也不可能涵盖所有的功能；如果有系统函数不支持的需求，我们就需要用自定义函数（User Defined Functions，UDF）来实现了。事实上，系统内置函数仍然在不断扩充，如果我们认为自己实现的自定义函数足够通用、应用非常广泛，也可以在项目跟踪工具 JIRA 上向 Flink 开发团队提出“议题”（issue），请求将新的函数添加到系统函数中。

2.1 编写自定义UDF函数

    自定义一个ScalarFunction，传入一个String类型的参数，输出这个参数的hashCode

public class HashScalarFunction extends ScalarFunction {    public String eval(String str){        return String.valueOf(str.hashCode());    }}

2.2 在代码中以SQL方式使用UDF函数

2.2.1 读取mysql数据使用UDF函数转换并输出到控制台

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");    }}

2.2.2 读取mysql数据使用UDF函数转换并输出到mysql

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);    }}

2.3 在StreamPark中以SQL方式使用UDF函数

在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为：

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;

运行作业后mysql可正常插入数据