Flink实战教程-自定义函数之TableFunction

大数据技术与应用实战

发布于 2020-09-15 14:16:42

4970

发布于 2020-09-15 14:16:42

文章被收录于专栏：大数据技术与应用实战大数据技术与应用实战

自定义函数
- 单个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维表功能，用来在流式数据中补全字段信息. 完整代码请参考：

https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/sql/function/CustomTableFunction.java

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-07-01，如有侵权请联系 cloudcommunity@tencent.com 删除

sql

flink

java

大数据

本文分享自大数据技术与应用实战微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度