背景 网站建设,建立网站需要多少钱八寇湖南岚鸿团队,建一个门户网站,石家庄网站建设需要多少钱Flink 系列文章
1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接
13、Flink 的table api与sql的基本概念、通用api介绍及入门示例 14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性 15、Flink 的ta…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 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之Catalogsjava api操作数据库、表-2 24、Flink 的table api与sql之Catalogsjava api操作视图-3 24、Flink 的table api与sql之Catalogsjava 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、DELETE1 29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE2 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 系列文章三、自定义函数1、概述2、开发指南1、函数类2、求值方法3、类型推导1、自动类型推导2、定制类型推导 4、确定性1、内置函数的确定性 5、运行时集成 3、标量函数-自定义函数说明及示例4、表值函数-自定义函数说明及示例 本文介绍了flink的自定义函数概述、开发指南以及标量函数、表值函数的自定义函数实现及说明提供的示例均可运行并提供运行结果供参考。 本文依赖flink集群能正常使用。 本文分为4个部分即自定义函数的概述、开发指南、标量自定义函数的说明及示例、表值自定义函数的说明及示例。 本文的示例均在Flink 1.17版本中运行。
三、自定义函数
自定义函数UDF是一种扩展开发机制可以用来在查询语句里调用难以用其他方式表达的频繁使用或自定义的逻辑。
自定义函数可以用 JVM 语言例如 Java 或 Scala或 Python 实现实现者可以在 UDF 中使用任意第三方库本文聚焦于使用 JVM 语言开发自定义函数。
1、概述
当前 Flink 有如下几种函数
标量函数将标量值转换成一个新标量值表值函数将标量值转换成新的行数据聚合函数将多行数据里的标量值转换成一个新标量值表值聚合函数将多行数据里的标量值转换成新的行数据异步表值函数是异步查询外部数据系统的特殊函数。 标量和表值函数已经使用了新的基于数据类型的类型系统聚合函数仍然使用基于 TypeInformation 的旧类型系统。 2、开发指南 在聚合函数使用新的类型系统前本节仅适用于标量和表值函数。 所有的自定义函数都遵循一些基本的实现原则。
1、函数类
实现类必须继承自合适的基类之一例如 org.apache.flink.table.functions.ScalarFunction 。
该类必须声明为 public 而不是 abstract 并且可以被全局访问。不允许使用非静态内部类或匿名类。
为了将自定义函数存储在持久化的 catalog 中该类必须具有默认构造器且在运行时可实例化。
Table API 中的匿名函数只有在函数不是有状态的stateful即仅包含瞬态和静态transient and static字段时才能持久化。
2、求值方法
基类提供了一组可以被重写的方法例如 open()、 close() 或 isDeterministic() 。
但是除了上述方法之外作用于每条传入记录的主要逻辑还必须通过专门的 求值方法 来实现。
根据函数的种类后台生成的运算符会在运行时调用诸如 eval()、accumulate() 或 retract() 之类的求值方法。
这些方法必须声明为 public 并带有一组定义明确的参数。
常规的 JVM 方法调用语义是适用的。因此可以
实现重载的方法例如 eval(Integer) 和 eval(LocalDateTime)使用变长参数例如 eval(Integer…);使用对象继承例如 eval(Object) 可接受 LocalDateTime 和 Integer 作为参数也可组合使用例如 eval(Object…) 可接受所有类型的参数。
示例片段
import org.apache.flink.table.functions.ScalarFunction;// 有多个重载求值方法的函数
public static class SumFunction extends ScalarFunction {//两Integer数求和public Integer eval(Integer a, Integer b) {return a b;}//两String数转换后求和public Integer eval(String a, String b) {return Integer.valueOf(a) Integer.valueOf(b);}//多Double数据求和public Integer eval(Double... d) {double result 0;for (double value : d)result value;return (int) result;}
}3、类型推导
Table类似于 SQL 标准是一种强类型的 API。因此函数的参数和返回类型都必须映射到数据类型。
从逻辑角度看Planner 需要知道数据类型、精度和小数位数从 JVM 角度来看Planner 在调用自定义函数时需要知道如何将内部数据结构表示为 JVM 对象。
术语 类型推导 概括了意在验证输入值、派生出参数/返回值数据类型的逻辑。
Flink 自定义函数实现了自动的类型推导提取通过反射从函数的类及其求值方法中派生数据类型。如果这种隐式的反射提取方法不成功则可以通过使用 DataTypeHint 和 FunctionHint 注解相关参数、类或方法来支持提取过程下面展示了有关如何注解函数的例子。
如果需要更高级的类型推导逻辑实现者可以在每个自定义函数中显式重写 getTypeInference() 方法。但是建议使用注解方式因为它可使自定义类型推导逻辑保持在受影响位置附近而在其他位置则保持默认状态。
1、自动类型推导
自动类型推导会检查函数的类和求值方法派生出函数参数和结果的数据类型 DataTypeHint 和 FunctionHint 注解支持自动类型推导。
有关可以隐式映射到数据类型的类的完整列表请参阅数据类型。
DataTypeHint
在许多情况下需要支持以 内联 方式自动提取出函数参数、返回值的类型。
以下例子展示了如何使用 DataTypeHint详情可参考该注解类的文档。
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;// 有多个重载求值方法的函数
public static class OverloadedFunction extends ScalarFunction {// no hint requiredpublic Long eval(long a, long b) {return a b;}// 定义 decimal 的精度和小数位public DataTypeHint(DECIMAL(12, 3)) BigDecimal eval(double a, double b) {return BigDecimal.valueOf(a b);}// 定义嵌套数据类型DataTypeHint(ROWs STRING, t TIMESTAMP_LTZ(3))public Row eval(int i) {return Row.of(String.valueOf(i), Instant.ofEpochSecond(i));}// 允许任意类型的符入并输出序列化定制后的值DataTypeHint(value RAW, bridgedTo ByteBuffer.class)public ByteBuffer eval(DataTypeHint(inputGroup InputGroup.ANY) Object o) {return MyUtils.serializeToByteBuffer(o);}
}FunctionHint
有时我们希望一种求值方法可以同时处理多种数据类型有时又要求对重载的多个求值方法仅声明一次通用的结果类型。
FunctionHint 注解可以提供从入参数据类型到结果数据类型的映射它可以在整个函数类或求值方法上注解输入、累加器和结果的数据类型。可以在类顶部声明一个或多个注解也可以为类的所有求值方法分别声明一个或多个注解。所有的 hint 参数都是可选的如果未定义参数则使用默认的基于反射的类型提取。在函数类顶部定义的 hint 参数被所有求值方法继承。
以下例子展示了如何使用 FunctionHint详情可参考该注解类的文档。
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;// 为函数类的所有求值方法指定同一个输出类型
FunctionHint(output DataTypeHint(ROWs STRING, i INT))
public static class OverloadedFunction extends TableFunctionRow {public void eval(int a, int b) {collect(Row.of(Sum, a b));}// overloading of arguments is still possiblepublic void eval() {collect(Row.of(Empty args, -1));}
}// 解耦类型推导与求值方法类型推导完全取决于 FunctionHint
FunctionHint(input {DataTypeHint(INT), DataTypeHint(INT)},output DataTypeHint(INT)
)
FunctionHint(input {DataTypeHint(BIGINT), DataTypeHint(BIGINT)},output DataTypeHint(BIGINT)
)
FunctionHint(input {},output DataTypeHint(BOOLEAN)
)
public static class OverloadedFunction extends TableFunctionObject {// an implementer just needs to make sure that a method exists that can be called by the JVMpublic void eval(Object... o) {if (o.length 0) {collect(false);}collect(o[0]);}
}2、定制类型推导
在大多数情况下DataTypeHint 和 FunctionHint 足以构建自定义函数然而通过重写 getTypeInference() 定制自动类型推导逻辑实现者可以创建任意像系统内置函数那样有用的函数。
以下用 Java 实现的例子展示了定制类型推导的潜力它根据字符串参数来确定函数的结果类型。该函数带有两个字符串参数第一个参数表示要分析的字符串第二个参数表示目标类型。
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.types.Row;public static class LiteralFunction extends ScalarFunction {public Object eval(String s, String type) {switch (type) {case INT:return Integer.valueOf(s);case DOUBLE:return Double.valueOf(s);case STRING:default:return s;}}// 禁用自动的反射式类型推导使用如下逻辑进行类型推导Overridepublic TypeInference getTypeInference(DataTypeFactory typeFactory) {return TypeInference.newBuilder()// 指定输入参数的类型必要时参数会被隐式转换.typedArguments(DataTypes.STRING(), DataTypes.STRING())// specify a strategy for the result data type of the function.outputTypeStrategy(callContext - {if (!callContext.isArgumentLiteral(1) || callContext.isArgumentNull(1)) {throw callContext.newValidationError(Literal expected for second argument.);}// 基于字符串值返回数据类型final String literal callContext.getArgumentValue(1, String.class).orElse(STRING);switch (literal) {case INT:return Optional.of(DataTypes.INT().notNull());case DOUBLE:return Optional.of(DataTypes.DOUBLE().notNull());case STRING:default:return Optional.of(DataTypes.STRING());}}).build();}
}4、确定性
每个用户自定义函数类都可以通过重写 isDeterministic() 方法来声明它是否产生确定性的结果。如果该函数不是纯粹函数式的如random(), date(), 或now()该方法必须返回 false。默认情况下isDeterministic() 返回 true。
此外重写 isDeterministic() 方法也可能影响运行时行为。运行时实现可能会在两个不同的阶段被调用 在生成执行计划期间如果一个函数是通过常量表达式调用的或者常量表达式可以从给定的语句中推导出来那么一个函数就会被预计算以减少常量表达式并且可能不再在集群上执行。 除非 isDeterministic() 被重写为 false 用来在这种情况下禁用常量表达式简化。比如说以下对 ABS 的调用在生成执行计划期间被执行SELECT ABS(-1) FROM t 和 SELECT ABS(field) FROM t WHERE field -1而 SELECT ABS(field) FROM t 则不执行。 在运行时即在集群执行如果一个函数被调用时带有非常量表达式或 isDeterministic() 返回 false。
1、内置函数的确定性
系统内置函数的确定性是不可改变的。存在两种不具有确定性的函数动态函数和非确定性函数根据 Apache Calcite SqlOperator 的定义 /*** Returns whether a call to this operator is guaranteed to always return* the same result given the same operands; true is assumed by default.*/public boolean isDeterministic() {return true;}/*** Returns whether it is unsafe to cache query plans referencing this* operator; false is assumed by default.*/public boolean isDynamicFunction() {return false;}isDeterministic 表示函数的确定性声明返回 false 时将在运行时对每个记录进行计算。 isDynamicFunction 声明返回 true 时意味着该函数只能在查询开始时被计算对于批处理模式它只在生成执行计划期间被执行 而对于流模式它等效于一个非确定性的函数这是因为查询在逻辑上是连续执行的流模式对动态表的连续查询抽象所以动态函数在每次查询执行时也会被重新计算当前实现下等效于每条记录计算。
以下内置函数总是非确定性的批和流模式下都在运行时对每条记录进行计算
UUIDRANDRAND_INTEGERCURRENT_DATABASEUNIX_TIMESTAMPCURRENT_ROW_TIMESTAMP
以下内置时间函数是动态的批处理模式下将在生成执行计划期间被执行查询开始对于流模式将在运行时对每条记录进行计算
CURRENT_DATECURRENT_TIMECURRENT_TIMESTAMPNOWLOCALTIMELOCALTIMESTAMP isDynamicFunction 仅适用于内置函数 5、运行时集成
有时候自定义函数需要获取一些全局信息或者在真正被调用之前做一些配置setup/清理clean-up的工作。自定义函数也提供了 open() 和 close() 方法你可以重写这两个方法做到类似于 DataStream API 中 RichFunction 的功能。
open() 方法在求值方法被调用之前先调用。close() 方法在求值方法调用完之后被调用。
open() 方法提供了一个 FunctionContext它包含了一些自定义函数被执行时的上下文信息比如 metric group、分布式文件缓存或者是全局的作业参数等。
下面的信息可以通过调用 FunctionContext 的对应的方法来获得
方法描述getMetricGroup()执行该函数的 subtask 的 Metric Group。getCachedFile(name)分布式文件缓存的本地临时文件副本。getJobParameter(name, defaultValue)跟对应的 key 关联的全局参数值。
下面的例子展示了如何在一个标量函数中通过 FunctionContext 来获取一个全局的任务参数
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;public static class HashCodeFunction extends ScalarFunction {private int factor 0;Overridepublic void open(FunctionContext context) throws Exception {// 获取参数 hashcode_factor// 如果不存在则使用默认值 12factor Integer.parseInt(context.getJobParameter(hashcode_factor, 12));}public int eval(String s) {return s.hashCode() * factor;}
}TableEnvironment env TableEnvironment.create(...);// 设置任务参数
env.getConfig().addJobParameter(hashcode_factor, 31);// 注册函数
env.createTemporarySystemFunction(hashCode, HashCodeFunction.class);// 调用函数
env.sqlQuery(SELECT myField, hashCode(myField) FROM MyTable);3、标量函数-自定义函数说明及示例
自定义标量函数可以把 0 到多个标量值映射成 1 个标量值数据类型里列出的任何数据类型都可作为求值方法的参数和返回值类型。
想要实现自定义标量函数你需要扩展 org.apache.flink.table.functions 里面的 ScalarFunction 并且实现一个或者多个求值方法。标量函数的行为取决于你写的求值方法。 求值方法必须是 public 的而且名字必须是 eval。 下面自定义函数是将balance加上万元以及求balance/age仅仅示例如何使用其运行结果在每次输出的代码后面注释的行。
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;
import java.util.List;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.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** author alanchan**/
public class TestUDScalarFunctionDemo {DataNoArgsConstructorAllArgsConstructorpublic static class User {private long id;private String name;private int age;private int balance;private Long rowtime;}final static ListUser userList Arrays.asList(new User(1L, alan, 18, 20,1698742358391L), new User(2L, alan, 19, 25,1698742359396L), new User(3L, alan, 25, 30,1698742360407L),new User(4L, alanchan, 28,35, 1698742361409L), new User(5L, alanchan, 29, 35,1698742362424L));public static class TestScalarFunction extends ScalarFunction {// 接受任意类型输入返回 String 型输出public String eval(DataTypeHint(inputGroup InputGroup.ANY) Object o) {return o.toString() (万元);}public double eval(Integer age, Integer balance) {return balance / age *1.0;}}/*** param args* throws Exception*/public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv StreamTableEnvironment.create(env);DataStreamUser users env.fromCollection(userList);Table usersTable tenv.fromDataStream(users, $(id), $(name), $(age),$(balance), $(rowtime));//1、 在 Table API 里不经注册直接“内联”调用函数Table result usersTable.select($(id), $(name), call(TestScalarFunction.class, $(balance)));DataStreamTuple2Boolean, Row resultDS tenv.toRetractStream(result, Row.class);
// resultDS.print();
// 11 (true,I[2, alan, 25 (万元)])
// 12 (true,I[3, alan, 30 (万元)])
// 13 (true,I[4, alanchan, 35 (万元)])
// 10 (true,I[1, alan, 20 (万元)])
// 14 (true,I[5, alanchan, 35 (万元)])Table result2 usersTable.select($(id), $(name), $(age), call(TestScalarFunction.class, $(balance)), call(TestScalarFunction.class, $(age), $(balance)));DataStreamTuple2Boolean, Row result2DS tenv.toRetractStream(result2, Row.class);
// result2DS.print();
// 9 (true,I[2, alan, 19, 25 (万元), 1.0])
// 10 (true,I[3, alan, 25, 30 (万元), 1.0])
// 12 (true,I[5, alanchan, 29, 35 (万元), 1.0])
// 11 (true,I[4, alanchan, 28, 35 (万元), 1.0])
// 8 (true,I[1, alan, 18, 20 (万元), 1.0])//2、 注册函数tenv.createTemporarySystemFunction(TestScalarFunction, TestScalarFunction.class);// 在 Table API 里调用注册好的函数Table result3 usersTable.select($(id), $(name),call(TestScalarFunction, $(balance)));DataStreamTuple2Boolean, Row result3DS tenv.toRetractStream(result3, Row.class);
// result3DS.print();
// 2 (true,I[4, alanchan, 35 (万元)])
// 3 (true,I[5, alanchan, 35 (万元)])
// 15 (true,I[1, alan, 20 (万元)])
// 16 (true,I[2, alan, 25 (万元)])
// 1 (true,I[3, alan, 30 (万元)])// 在 SQL 里调用注册好的函数tenv.createTemporaryView(user_view, users);Table result4 tenv.sqlQuery(SELECT id,name,TestScalarFunction(balance) ,TestScalarFunction(age,balance) FROM user_view);DataStreamTuple2Boolean, Row result4DS tenv.toRetractStream(result4, Row.class);result4DS.print();
// 14 (true,I[1, alan, 20 (万元), 1.0])
// 1 (true,I[4, alanchan, 35 (万元), 1.0])
// 2 (true,I[5, alanchan, 35 (万元), 1.0])
// 15 (true,I[2, alan, 25 (万元), 1.0])
// 16 (true,I[3, alan, 30 (万元), 1.0])env.execute();}}4、表值函数-自定义函数说明及示例
跟自定义标量函数一样自定义表值函数的输入参数也可以是 0 到多个标量。但是跟标量函数只能返回一个值不同的是它可以返回任意多行。返回的每一行可以包含 1 到多列如果输出行只包含 1 列会省略结构化信息并生成标量值这个标量值在运行阶段会隐式地包装进行里。
要定义一个表值函数你需要扩展 org.apache.flink.table.functions 下的 TableFunction可以通过实现多个名为 eval 的方法对求值方法进行重载。像其他函数一样输入和输出类型也可以通过反射自动提取出来。表值函数返回的表的类型取决于 TableFunction 类的泛型参数 T不同于标量函数表值函数的求值方法本身不包含返回类型而是通过 collect(T) 方法来发送要输出的行。
在 Table API 中表值函数是通过 .joinLateral(…) 或者 .leftOuterJoinLateral(…) 来使用的。joinLateral 算子会把外表算子左侧的表的每一行跟跟表值函数返回的所有行位于算子右侧进行 crossjoin。leftOuterJoinLateral 算子也是把外表算子左侧的表的每一行跟表值函数返回的所有行位于算子右侧进行crossjoin并且如果表值函数返回 0 行也会保留外表的这一行。
在 SQL 里面用 JOIN 或者 以 ON TRUE 为条件的 LEFT JOIN 来配合 LATERAL TABLE() 的使用。
下面示例中包含表值函数的四种应用方式。
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;
import java.util.List;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 TestUDTableFunctionDemo {DataNoArgsConstructorAllArgsConstructorpublic static class User {private long id;private String name;private int age;private int balance;private Long rowtime;}final static ListUser userList Arrays.asList(new User(1L, alan,chen, 18, 20,1698742358391L), new User(2L, alan,chen, 19, 25,1698742359396L), new User(3L, alan,chen, 25, 30,1698742360407L),new User(4L, alan,chan, 28,35, 1698742361409L), new User(5L, alan,chan, 29, 35,1698742362424L));FunctionHint(output DataTypeHint(ROWfirstName STRING, lastName String))public static class SplitFunction extends TableFunctionRow {public void eval(String str) {String[] names str.split(,);collect(Row.of(names[0],names[1]));
// for (String s : str.split(, )) {
// // use collect(...) to emit a row
// collect(Row.of(s, s.length()));
// }}}FunctionHint(output DataTypeHint(ROWid int, name String, age int, balance int, rowtime string))public static class OverloadedFunction extends TableFunctionRow {public void eval(String str) {String[] user str.split(,);collect(Row.of(Integer.valueOf(user[0]),user[1],Integer.valueOf(user[2]),Integer.valueOf(user[3]),user[4]));}}/*** param args* throws Exception */public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv StreamTableEnvironment.create(env);DataStreamUser users env.fromCollection(userList);Table usersTable tenv.fromDataStream(users, $(id), $(name), $(age), $(balance), $(rowtime));// 1、 在 Table API 里不经注册直接“内联”调用函数Table result usersTable.joinLateral(call(SplitFunction.class, $(name))).select($(id), $(name),$(firstName),$(lastName));DataStreamTuple2Boolean, Row resultDS tenv.toRetractStream(result, Row.class);
// resultDS.print();
// 11 (true,I[5, alan,chan, alan, chan])
// 7 (true,I[1, alan,chen, alan, chen])
// 9 (true,I[3, alan,chen, alan, chen])
// 10 (true,I[4, alan,chan, alan, chan])
// 8 (true,I[2, alan,chen, alan, chen])DataStreamString 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));Table result2 usersTable2.joinLateral(call(OverloadedFunction.class, $(userString))).select($(userString),$(id),$(name),$(age),$(balance),$(rowtime)) ; DataStreamTuple2Boolean, Row result2DS tenv.toRetractStream(result2, Row.class);
// result2DS.print();
// 15 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 13 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 14 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 11 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 12 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])Table result3 usersTable2.leftOuterJoinLateral(call(OverloadedFunction.class, $(userString))).select($(userString),$(id),$(name),$(age),$(balance),$(rowtime)) ; DataStreamTuple2Boolean, Row result3DS tenv.toRetractStream(result3, Row.class);
// result3DS.print();
// 5 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 6 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 3 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 4 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
// 7 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])// 在 Table API 里重命名函数字段Table result4 usersTable2.leftOuterJoinLateral(call(OverloadedFunction.class, $(userString)).as(t_id,t_name,t_age,t_balance,t_rowtime)).select($(userString),$(t_id),$(t_name),$(t_age),$(t_balance),$(t_rowtime)) ; DataStreamTuple2Boolean, Row result4DS tenv.toRetractStream(result4, Row.class);
// result4DS.print();
// 10 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 13 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 14 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 12 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 11 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])//2、 注册函数tenv.createTemporarySystemFunction(OverloadedFunction, OverloadedFunction.class);// 在 Table API 里调用注册好的函数Table result5 usersTable2.leftOuterJoinLateral(call(OverloadedFunction, $(userString)).as(t_id,t_name,t_age,t_balance,t_rowtime)).select($(userString),$(t_id),$(t_name),$(t_age),$(t_balance),$(t_rowtime)) ; DataStreamTuple2Boolean, Row result5DS tenv.toRetractStream(result5, Row.class);
// result5DS.print();
// 11 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 14 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 15 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 13 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 12 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])Table result6 usersTable2.joinLateral(call(OverloadedFunction, $(userString)).as(t_id,t_name,t_age,t_balance,t_rowtime)).select($(userString),$(t_id),$(t_name),$(t_age),$(t_balance),$(t_rowtime)) ; DataStreamTuple2Boolean, Row result6DS tenv.toRetractStream(result6, Row.class);
// result6DS.print();
// 8 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 9 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 5 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 7 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 6 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])//3、 在 SQL 里调用注册好的函数tenv.createTemporaryView(user_view, usersTable2);Table result7 tenv.sqlQuery(SELECT userString, id,name,age,balance,rowtime FROM user_view, LATERAL TABLE(OverloadedFunction(userString)));DataStreamTuple2Boolean, Row result7DS tenv.toRetractStream(result7, Row.class);
// result7DS.print();
// 15 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 13 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 1 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 14 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
// 16 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])Table result8 tenv.sqlQuery(SELECT userString, id,name,age,balance,rowtime FROM user_view LEFT JOIN LATERAL TABLE( OverloadedFunction(userString)) ON TRUE );DataStreamTuple2Boolean, Row result8DS tenv.toRetractStream(result8, Row.class);
// result8DS.print();
// 13 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
// 1 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 15 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 14 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
// 16 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])//4、 在 SQL 里重命名函数字段Table result9 tenv.sqlQuery(SELECT userString, t_id, t_name,t_age,t_balance,t_rowtime FROM user_view LEFT JOIN LATERAL TABLE(OverloadedFunction(userString)) AS T(t_id, t_name,t_age,t_balance,t_rowtime) ON TRUE);DataStreamTuple2Boolean, Row result9DS tenv.toRetractStream(result9, Row.class);result9DS.print();
// 7 (true,I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
// 10 (true,I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
// 9 (true,I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
// 8 (true,I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
// 6 (true,I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])env.execute();}}以上介绍了flink的自定义函数概述、开发指南以及标量函数、表值函数的自定义函数实现及说明提供的示例均可运行并提供运行结果供参考。
