目录：
- 自定义函数
- 单个eval方法
- 多个eval方法
- 不固定参数
- 通过注解指定返回类型
- 注册函数
- 构造数据源
- 查询
- left join
- join
- 多种类型参数
- 不固定参数类型
今天我们来聊聊flink sql中另外一种自定义函数-TableFuntion. TableFuntion 可以有0个、一个、多个输入参数，他的返回值可以是任意行，每行可以有多列数据.
实现自定义TableFunction需要继承TableFunction类，然后定义一个public类型的eval方法。结合官网的例子具体来讲解一下。
自定义函数
单个eval方法
代码语言：javascript复制 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方法
代码语言：javascript复制 /**
* 注册多个eval方法，接收long或者string类型的参数，然后将他们转成string类型
*/
public static class DuplicatorFunction extends TableFunction<String>{
public void eval(Long i){
eval(String.valueOf(i));
}
public void eval(String s){
collect(s);
}
}不固定参数
代码语言：javascript复制 /**
* 接收不固定个数的int型参数,然后将所有数据依次返回
*/
public static class FlattenFunction extends TableFunction<Integer>{
public void eval(Integer... args){
for (Integer i: args){
collect(i);
}
}
}通过注解指定返回类型
代码语言：javascript复制 /**
* 通过注册指定返回值类型，flink 1.11 版本开始支持
*/
@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，然后把上述三个函数都注册了
代码语言：javascript复制 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());构造数据源
代码语言：javascript复制 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
代码语言：javascript复制 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
代码语言：javascript复制 String sql = "SELECT o.currency, T.word, T.length FROM Orders as o ,"
" LATERAL TABLE(split(currency)) as T(word, length)";多种类型参数
代码语言：javascript复制 String sql2 = "SELECT * FROM Orders as o , "
"LATERAL TABLE(duplicator(amount)),"
"LATERAL TABLE(duplicator(currency))";不固定参数类型
代码语言：javascript复制 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
