19、Flink 的Table API 和 SQL 中的自定义函数及示例（4）

Flink 系列文章

1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接

13、Flink 的table api与sql的基本概念、通用api介绍及入门示例
14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性
15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例（1）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例（2）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例（3）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例（4）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例（6）
17、Flink 之Table API: Table API 支持的操作（1）
17、Flink 之Table API: Table API 支持的操作（2）
18、Flink的SQL 支持的操作和语法
19、Flink 的Table API 和 SQL 中的内置函数及示例（1）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（2）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（3）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（4）
20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL，可以直接提交 SQL 任务到集群上

22、Flink 的table api与sql之创建表的DDL
24、Flink 的table api与sql之Catalogs（介绍、类型、java api和sql实现ddl、java api和sql操作catalog）-1
24、Flink 的table api与sql之Catalogs（java api操作数据库、表）-2
24、Flink 的table api与sql之Catalogs（java api操作视图）-3
24、Flink 的table api与sql之Catalogs（java api操作分区与函数）-4

26、Flink 的SQL之概览与入门示例
27、Flink 的SQL之SELECT (select、where、distinct、order by、limit、集合操作和去重)介绍及详细示例（1）
27、Flink 的SQL之SELECT (SQL Hints 和 Joins)介绍及详细示例（2）
27、Flink 的SQL之SELECT (窗口函数)介绍及详细示例（3）
27、Flink 的SQL之SELECT (窗口聚合)介绍及详细示例（4）
27、Flink 的SQL之SELECT (Group Aggregation分组聚合、Over Aggregation Over聚合和 Window Join 窗口关联)介绍及详细示例（5）
27、Flink 的SQL之SELECT (Top-N、Window Top-N 窗口 Top-N 和 Window Deduplication 窗口去重)介绍及详细示例（6）
27、Flink 的SQL之SELECT (Pattern Recognition 模式检测)介绍及详细示例（7）
28、Flink 的SQL之DROP 、ALTER 、INSERT 、ANALYZE 语句
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（1）
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（2）
30、Flink SQL之SQL 客户端（通过kafka和filesystem的例子介绍了配置文件使用-表、视图等）
32、Flink table api和SQL 之用户自定义 Sources & Sinks实现及详细示例
41、Flink之Hive 方言介绍及详细示例
42、Flink 的table api与sql之Hive Catalog
43、Flink之Hive 读写及详细验证示例
44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例–网上有些说法好像是错误的

文章目录

Flink 系列文章
- 7、sql clinet中应用自定义函数
- - 1）、实现自定义函数
  - 2）、打包并上传jar至flink的lib目录下
  - 3）、验证
  - - 1、创建表
    - 2、初始化表数据
    - 3、注册函数
    - 4、验证自定义函数
- 8、pojo 数据类型应用示例-表值函数

本文展示了自定义函数在Flink sql client的应用以及自定义函数中使用pojo的示例。
本文依赖flink、kafka集群能正常使用。
本文分为2个部分，即自定义函数在Flink sql client中的应用以及自定义函数中使用pojo数据类型。
本文的示例如无特殊说明则是在Flink 1.17版本中运行。

7、sql clinet中应用自定义函数

本示例将上文中自定义的函数打包后在flink sql client中进行应用。

1）、实现自定义函数

本文的所有示例需要依赖的maven见本篇的上一篇：17、Flink 之Table API: Table API 支持的操作（1）
或者引入

    <!-- flink依赖引入--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>${flink.version}</version><scope>provided</scope></dependency>

示例代码

package org.table_sql;import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;/*** @author alanchan**/@FunctionHint(output = @DataTypeHint("ROW<id int, name String, age int, balance int, rowtime string>"))
public class Alan_SplitFunction extends TableFunction<Row> {public void eval(String str) {String[] row = str.split(",");collect(Row.of(Integer.valueOf(row[0]), row[1], Integer.valueOf(row[2]), Integer.valueOf(row[3]), row[4]));}}

2）、打包并上传jar至flink的lib目录下

将该文件打包成jar文件，特别说明的是，注意flink运行环境与打包引入的jar文件是否冲突，推荐做法是只打包创建自定义函数所依赖的jar文件，其他jar使用flink部署环境的jar。
本示例打包后的文件名：Alan_SplitFunction.jar
上传jar文件后，并重启flink集群。

3）、验证

1、创建表

Flink SQL> SET sql-client.execution.result-mode = tableau;
[INFO] Session property has been set.Flink SQL> CREATE TABLE alan_split_table (
>   userString STRING
> ) WITH (
>   'connector' = 'kafka',
>   'topic' = 'alan_split',
>   'properties.bootstrap.servers' = '192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092',
>   'properties.group.id' = 'testGroup',
>   'scan.startup.mode' = 'earliest-offset',
>   'format' = 'csv'
> );
[INFO] Execute statement succeed.Flink SQL> select * from alan_split_table;
[INFO] Result retrieval cancelled.

2、初始化表数据

本示例是通过kafka队列插入的数据，前提是kafka环境好用。

[alanchan@server1 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic alan_split
>"11,alan,18,20,1699341167461"
>"12,alan,19,25,1699341168464"
>"13,alan,20,30,1699341169472"
>"14,alanchan,18,22,1699341170479"
>"15,alanchan,19,25,1699341171482"Flink SQL> select * from alan_split_table;
+----+--------------------------------+
| op |                     userString |
+----+--------------------------------+
| +I |    11,alan,18,20,1699341167461 |
| +I |    12,alan,19,25,1699341168464 |
| +I |    13,alan,20,30,1699341169472 |
| +I | 14,alanchan,18,22,169934117... |
| +I | 15,alanchan,19,25,169934117... |

3、注册函数

将自定义的函数注册为flink的临时函数，临时函数只在当前的会话中起作用，如果注册成其他函数，参考如下语法

CREATE [TEMPORARY|TEMPORARY SYSTEM] FUNCTION[IF NOT EXISTS] [[catalog_name.]db_name.]function_nameAS identifier [LANGUAGE JAVA|SCALA|PYTHON][USING JAR '<path_to_filename>.jar' [, JAR '<path_to_filename>.jar']* ]# TEMPORARY
# 创建一个有 catalog 和数据库命名空间的临时 catalog function ，并覆盖原有的 catalog function 。# TEMPORARY SYSTEM
# 创建一个没有数据库命名空间的临时系统 catalog function ，并覆盖系统内置的函数。

本示例注册为临时函数，如下

Flink SQL> CREATE TEMPORARY FUNCTION alan_split AS 'org.table_sql.Alan_SplitFunction';
[INFO] Execute statement succeed.Flink SQL> show functions;
+-----------------------+
|         function name |
+-----------------------+
|                IFNULL |
|      SOURCE_WATERMARK |
|                TYPEOF |
|                   abs |
|                  acos |
|            alan_split |
|                   and |
|                 array |
。。。。。。

4、验证自定义函数

Flink SQL> SELECT userString, t_id, t_name,t_age,t_balance,t_rowtime 
> FROM alan_split_table 
> LEFT JOIN LATERAL TABLE(alan_split(userString)) AS T(t_id, t_name,t_age,t_balance,t_rowtime) ON TRUE;
+----+--------------------------------+-------------+--------------------------------+-------------+-------------+--------------------------------+
| op |                     userString |        t_id |                         t_name |       t_age |   t_balance |                      t_rowtime |
+----+--------------------------------+-------------+--------------------------------+-------------+-------------+--------------------------------+
| +I |    11,alan,18,20,1699341167461 |          11 |                           alan |          18 |          20 |                  1699341167461 |
| +I |    12,alan,19,25,1699341168464 |          12 |                           alan |          19 |          25 |                  1699341168464 |
| +I |    13,alan,20,30,1699341169472 |          13 |                           alan |          20 |          30 |                  1699341169472 |
| +I | 14,alanchan,18,22,169934117... |          14 |                       alanchan |          18 |          22 |                  1699341170479 |
| +I | 15,alanchan,19,25,169934117... |          15 |                       alanchan |          19 |          25 |                  1699341171482 |

至此，完成了自定义函数注册至flink sql client的验证。

8、pojo 数据类型应用示例-表值函数

功能参考 19、Flink 的Table API 和 SQL 中的自定义函数及示例（2）中的【4、表值函数-自定义函数说明及示例】
本示例仅仅是展示在自定义函数中使用pojo 对象。

本示例仅仅是一种实现方式，也可以覆盖getTypeInference并以编程方式提供所有组件，不再赘述。

本示例仅仅是以表值函数作为示例，其他的自定义函数类似。

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** @author alanchan**/
public class TestUDTableFunctionDemo2 {@Data@NoArgsConstructor@AllArgsConstructorpublic static class User {private int id;private String name;private int age;private int balance;private String rowtime;}//	@FunctionHint(output = @DataTypeHint("User<id int, name String, age int, balance int, rowtime string>"))
//	public static class OverloadedFunction extends TableFunction<Row> {@FunctionHint(output =@DataTypeHint(bridgedTo = User.class))public static class OverloadedFunction extends TableFunction<User> {public void eval(String str) {String[] user = str.split(",");// 使用 Row数据类型
//			collect(Row.of(Integer.valueOf(user[0]), user[1], Integer.valueOf(user[2]), Integer.valueOf(user[3]), user[4]));// 使用User pojo数据类型collect(new User(Integer.valueOf(user[0]), user[1], Integer.valueOf(user[2]), Integer.valueOf(user[3]), user[4]));}}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv = StreamTableEnvironment.create(env);DataStream<String> row = env.fromCollection(//id name age balance rowtimeArrays.asList("11,alan,18,20,1699341167461","12,alan,19,25,1699341168464","13,alan,20,30,1699341169472","14,alanchan,18,22,1699341170479","15,alanchan,19,25,1699341171482"));Table usersTable2 = tenv.fromDataStream(row, $("userString"));tenv.createTemporarySystemFunction("OverloadedFunction", OverloadedFunction.class);Table result5 = usersTable2.leftOuterJoinLateral(call("OverloadedFunction", $("userString")).as("t_id","t_name","t_age","t_balance","t_rowtime")).select($("t_id"),$("t_name"),$("t_age"),$("t_balance"),$("t_rowtime")
//				.select($("userString"),$("t_id"),$("t_name"),$("t_age"),$("t_balance"),$("t_rowtime"))	;	DataStream<Tuple2<Boolean, Row>> result5DS = tenv.toRetractStream(result5, Row.class);result5DS.print();
//		15> (true,+I[15, alanchan, 19, 25, 1699341171482])
//		12> (true,+I[12, alan, 19, 25, 1699341168464])
//		13> (true,+I[13, alan, 20, 30, 1699341169472])
//		11> (true,+I[11, alan, 18, 20, 1699341167461])
//		14> (true,+I[14, alanchan, 18, 22, 1699341170479])env.execute();}}