文章目录
自定义函数单个eval方法多个eval方法不固定参数通过注解指定返回类型
注册函数构造数据源查询left joinjoin多种类型参数不固定参数类型
今天我们来聊聊flink sql中另外一种自定义函数-TableFuntion.
TableFuntion 可以有0个、一个、多个输入参数，他的返回值可以是任意行，每行可以有多列数据.
实现自定义TableFunction需要继承TableFunction类，然后定义一个public类型的eval方法。结合官网的例子具体来讲解一下。
自定义函数
单个eval方法
public static class Split extends TableFunction<Tuple2
<String,Integer>> {
private String separator
= ",";
public Split(String separator
) {
this.separator
= separator
;
}
public void eval(String str
) {
for (String s
: str
.split(separator
)) {
collect(new Tuple2<String,Integer>(s
, s
.length()));
}
}
}
来解释一下：
这个函数接收一个字符串类型的入参,将传进来的字符串用指定分隔符拆分，然后返回值是一组Tuple2，每个Tuple2包含单词以及其长度.TableFunction是一个泛型类，需要指定返回值类型不同于标量函数，eval方法没有返回值，使用collect方法来收集对象。
多个eval方法
public static class DuplicatorFunction extends TableFunction<String>{
public void eval(Long i
){
eval(String
.valueOf(i
));
}
public void eval(String s
){
collect(s
);
}
}
不固定参数
public static class FlattenFunction extends TableFunction<Integer>{
public void eval(Integer
... args
){
for (Integer i
: args
){
collect(i
);
}
}
}
通过注解指定返回类型
@FunctionHint(output
= @DataTypeHint("ROW< i INT, s STRING >"))
class DuplicatorFunction extends TableFunction<Row> {
public void eval(Integer i
, String s
) {
collect(Row
.of(i
, s
));
collect(Row
.of(i
, s
));
}
}
注册函数
这里使用blink的planner，然后把上述三个函数都注册了
StreamExecutionEnvironment env
= StreamExecutionEnvironment
.getExecutionEnvironment();
EnvironmentSettings bsSettings
= EnvironmentSettings
.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
StreamTableEnvironment tEnv
= StreamTableEnvironment
.create(env
, bsSettings
);
tEnv
.registerFunction("split", new Split(" "));
tEnv
.registerFunction("duplicator", new DuplicatorFunction());
tEnv
.registerFunction("flatten", new FlattenFunction());
构造数据源
List
<Tuple2
<Long,String>> ordersData
= new ArrayList<>();
ordersData
.add(Tuple2
.of(2L
, "Euro"));
ordersData
.add(Tuple2
.of(1L
, "US Dollar"));
ordersData
.add(Tuple2
.of(50L
, "Yen"));
ordersData
.add(Tuple2
.of(3L
, "Euro"));
DataStream
<Tuple2
<Long,String>> ordersDataStream
= env
.fromCollection(ordersData
);
Table orders
= tEnv
.fromDataStream(ordersDataStream
, "amount, currency, proctime.proctime");
tEnv
.registerTable("Orders", orders
);
查询
left join
Table result
= tEnv
.sqlQuery(
"SELECT o.currency, T.word, T.length FROM Orders as o LEFT JOIN LATERAL TABLE(split(currency)) as T(word, length) ON TRUE");
tEnv
.toAppendStream(result
, Row
.class).print();
解释一下：
有两种使用方式， 使用 join的时候用LATERAL TABLE ，使用left join的时候用LATERAL TABLE … ON TRUE.给TableFuntion返回的数据起一个别名：T(word, length)，其中T是表的别名，word和length是字段别名,所以我们可以在sql中通过o.currency, T.word, T.length来查询字段。
join
String sql
= "SELECT o.currency, T.word, T.length FROM Orders as o ," +
" LATERAL TABLE(split(currency)) as T(word, length)";
多种类型参数
String sql2
= "SELECT * FROM Orders as o , " +
"LATERAL TABLE(duplicator(amount))," +
"LATERAL TABLE(duplicator(currency))";
不固定参数类型
String sql3
= "SELECT * FROM Orders as o , " +
"LATERAL TABLE(flatten(100,200,300))";
今天这个TableFuntion我们就先讲到这里，后续我们通过自定义的TableFuntion来实现一个mysql维表和hbase维表功能，用来在流式数据中补全字段信息. 完整代码请参考：
https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/sql/function/CustomTableFunction.java
更多精彩内容，欢迎关注我的公众号【大数据技术与应用实战】！
