目录：
自定义函数
单个eval方法
多个eval方法
不固定参数
通过注解指定返回类型
注册函数
构造数据源
查询
left join
join
多种类型参数
不固定参数类型
今天我们来聊聊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方法
/** * 注册多个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); } }
不固定参数
/** * 接收不固定个数的int型参数,然后将所有数据依次返回 */ public static class FlattenFunction extends TableFunction<Integer>{ public void eval(Integer... args){ for (Integer i: args){ collect(i); } } }
通过注解指定返回类型
/** * 通过注册指定返回值类型，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，然后把上述三个函数都注册了
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维表功能，用来在流式数据中补全字段信息. 完整代码请参考：
更多精彩内容，欢迎关注我的公众号【大数据技术与应用实战】！
本文分享自微信公众号 - 大数据技术与应用实战（bigdata_bigdata）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
