内置函数

参考官网: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/functions/systemFunctions.html

自定义函数

自定义函数分类:
1) 标量函数(Scalar functions) 将标量值转换成一个新标量值；
2) 表值函数(Table functions) 将标量值转换成新的行数据；
3) 聚合函数(Aggregate functions) 将多行数据里的标量值转换成一个新标量值；
4) 表值聚合函数(Table aggregate) 将多行数据里的标量值转换成新的行数据；
异步表值函数(Async table functions) 是异步查询外部数据系统的特殊函数。

标量函数

可以将0、1或多个标量值，映射到新的标量值

TableAPI

  1.         // 1. 在tableApi中使用自定义函数
  2.         // 1.1 内联方式方式使用
  3.         table
  4.             .select($("f0"), call(MyUpperCase.class, $("f0")).as("u"))
  5.             .execute()
  6.             .print();
  7.         // 1.2 注册之后使用
  8.         tEnv.createTemporaryFunction("upper", MyUpperCase.class);
  9.         table
  10.             .select($("f0"), call("upper", $("f0")).as("u"))
  11.             .execute()
  12.             .print();

SQLAPI

  1.                 // 2. 在sql中使用
  2.         // 只能先注册再使用
  3.         tEnv.createTemporaryFunction("upper", MyUpperCase.class);
  4.         tEnv.sqlQuery("select f0, upper(f0) u from " + table).execute().print();
  1. package com.atguigu.chapter11_func_tableapi_sql.function;
  3. import org.apache.flink.configuration.Configuration;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.Table;
  6. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. import org.apache.flink.table.functions.ScalarFunction;
  9. /**
  10.  * @Author lizhenchao@atguigu.cn
  11.  * @Date 2021/7/26 9:11
  12.  */
  13. public class Flink01_Function_Scala_01 {
  14.     public static void main(String[] args) {
  15.         Configuration conf = new Configuration();
  16.         conf.setInteger("rest.port", 20000);
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  18.         env.setParallelism(1);
  20.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  22.         Table table = tEnv.fromValues("hello", "atguigu", "abc");
  24.         // 1. 在tableApi中使用自定义函数
  25.         // 1.1 内联方式方式使用
  26.         /*table
  27.             .select($("f0"), call(MyUpperCase.class, $("f0")).as("u"))
  28.             .execute()
  29.             .print();*/
  30.         // 1.2 注册之后使用
  31.         /*tEnv.createTemporaryFunction("upper", MyUpperCase.class);
  32.         table
  33.             .select($("f0"), call("upper", $("f0")).as("u"))
  34.             .execute()
  35.             .print();*/
  36.         // 2. 在sql中使用
  37.         // 只能先注册再使用
  38.         tEnv.createTemporaryFunction("upper", MyUpperCase.class);
  39.         tEnv.sqlQuery("select f0, upper(f0) u from " + table).execute().print();
  42.     }
  43.     // 函数
  45.     public static class MyUpperCase extends ScalarFunction {
  46.         public String eval(String s) {
  47.             if (s == null) {
  48.                 return null;
  49.             }
  50.             return s.toUpperCase();
  51.         }
  52.     }
  53. }

表值函数

看不懂就看视频吧

函数

  1. // row 是弱类型，需要使用注释告诉row内有什么类型
  2.     @FunctionHint(output = @DataTypeHint("row<w string, len int>"))
  3.     public static class SplitAndLen extends TableFunction<Row> {
  4.         public void eval(String s) {
  5.             if (s.contains("hello")) return(打断程序，不让collect写出则为null);// 如果是内连接，只有abc a b c，如果是左外连接 ，左边f0全包含，右边有hello的项都不包含
  6.             String[] words = s.split(" ");
  8.             for (String word : words) {
  9.                 // 调用几次就生成几行
  10.                 collect(Row.of(word, word.length()));
  11.             }
  12.         }
  13.     }

TableAPI

  1. // 1. table
  2.         // 内联
  3.         table
  4.             //.joinLateral(call(SplitAndLen.class, $("f0")))  // 内连接
  5.             .leftOuterJoinLateral(call(SplitAndLen.class, $("f0")))  // 左外连接
  6.             .select($("f0"), $("w"), $("len"))
  7.             .execute()
  8.             .print();
  9.           -----------------------------------------------------------------------
  10.         // 注册
  11.         tEnv.createTemporaryFunction("split", SplitAndLen.class);
  12.         table
  13.               //   .joinLateral(call("split", $("f0")))  // 内连接
  14.             .leftOuterJoinLateral(call(SplitAndLen.class, $("f0")))  // 左外连接
  15.             .select($("f0"), $("w"), $("len"))
  16.             .execute()
  17.             .print();

SQLAPI

  1. // 2. sql
  2.         tEnv.createTemporaryView("t1", table);
  4.         tEnv.createTemporaryFunction("split", SplitAndLen.class);
  6.         // 内连接的写法1:
  7.         tEnv.sqlQuery("select " +
  8.                           " f0," +
  9.                           " w," +
  10.                           " len " +
  11.                           "from t1 " +
  12.                           "join lateral table(split(f0)) on true")、//这个on是因为有join所以一定要有on
  13.                          // "left join lateral table(split(f0)) on true")// 下面的写法不能left join
  14.             .execute()
  15.             .print();
  16.         ------------------------------------------------------------
  17.         // select ... from a,b where a.id=b.id; ====  select .. from a join b on a.id=b.id
  18.         // 内连接的写法2:
  19.         tEnv.sqlQuery("select " +
  20.                           " f0," +
  21.                           " w1," +
  22.                           " len1 " +
  23.                           "from t1, " +
  24.                           " lateral table(split(f0)) as t(w1,len1)") //不能用外连接，as t(w1,len1)可以不写
  25.             .execute()
  26.             .print();

结果

sqlAPI 自动编码为upsert 只看最终结果即可
image.png

聚合函数

比较特别的地方是函数重写两个方法，调用时和普通聚合函数一样，group by 即可

函数

  1.  public static class MyAcc{
  2.         public Integer sum = 0;
  3.     }
  5.     public static class MySum extends AggregateFunction<Integer, MyAcc>{
  7.         // 返回最终聚合的结果
  8.         @Override
  9.         public Integer getValue(MyAcc acc) {
  10.             return acc.sum;
  11.         }
  12.         // 初始化累加器
  13.         @Override
  14.         public MyAcc createAccumulator() {
  15.             return new MyAcc();
  16.         }
  18.         // 累加
  19.         public void accumulate(MyAcc acc, Integer v){
  20.             acc.sum += v;
  21.         }
  22.     }

TableAPI

  1. tEnv.createTemporaryFunction("my_sum", MySum.class);
  2. table
  3.             .groupBy($("id"))
  4.             .select($("id"), call("my_sum", $("vc")).as("vc_sum"))
  5.             .execute()
  6.             .print();

SQLAPI

  1. tEnv.createTemporaryFunction("my_sum", MySum.class);
  2. tEnv.sqlQuery("select id, my_sum(vc) sum_vc from " + table + " group by id").execute().print();

表值聚合函数

又有聚合，又有制表，典型例子是topN

函数

  1. public static class FirstSecond {
  2.         public Integer first = 0;
  3.         public Integer second = 0;
  4.     }
  7. // 普通单值类型demo
  8. public static class Top2 extends TableAggregateFunction<String, FirstSecond> {
  10.         @Override
  11.         public FirstSecond createAccumulator() {
  12.             return new FirstSecond();
  13.         }
  15.         // 聚合函数
  16.         public void accumulate(FirstSecond acc, Integer v) {
  17.             if (v > acc.first) {
  18.                 acc.second = acc.first;
  19.                 acc.first = v;
  20.             } else if (v > acc.second) {
  21.                 acc.second = v;
  22.             }
  23.         }
  25.         // 指标
  26.         public void emitValue(FirstSecond acc, Collector<String> collector) {
  27.             collector.collect("最大水位: " + acc.first);  // 调用几次就有几行
  28.             collector.collect("第二大水位: " + acc.second);  // 调用几次就有几行
  29.         }
  30.     }
  32.     -------------------------------------
  34. //    元祖类型输出demo（pojo也是多列）
  35.  public static class Top22 extends TableAggregateFunction<Tuple2<Integer, Integer>, FirstSecond> {
  37.         @Override
  38.         public FirstSecond createAccumulator() {
  39.             return new FirstSecond();
  40.         }
  42.         // 聚合函数
  43.         public void accumulate(FirstSecond acc, Integer v) {
  44.             if (v > acc.first) {
  45.                 acc.second = acc.first;
  46.                 acc.first = v;
  47.             } else if (v > acc.second) {
  48.                 acc.second = v;
  49.             }
  50.         }
  52.         // 制表
  53.         public void emitValue(FirstSecond acc, Collector<Tuple2<Integer, Integer>> collector) {
  54.             collector.collect(Tuple2.of(acc.first, 1));  // 调用几次就有几行
  56.             if (acc.second > 0)
  57.                 collector.collect(Tuple2.of(acc.second, 2));  // 调用几次就有几行
  58.         }
  59.     }

TableAPI

  1. public static void main(String[] args) {
  2.         Configuration conf = new Configuration();
  3.         conf.setInteger("rest.port", 20000);
  4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  5.         env.setParallelism(1);
  7.         DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
  8.                                                                 new WaterSensor("sensor_1", 2000L, 20),
  9.                                                                 new WaterSensor("sensor_2", 3000L, 30),
  10.                                                                 new WaterSensor("sensor_1", 4000L, 40),
  11.                                                                 new WaterSensor("sensor_1", 5000L, 50),
  12.                                                                 new WaterSensor("sensor_2", 6000L, 60));
  14.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  16.         Table table = tEnv.fromDataStream(stream);
  18.         // 接下来我们写一个自定义TableAggregateFunction，用来提取每个sensor最高的两个温度值(两行)。
  19.         table
  20.             .groupBy($("id"))
  21. //            .flatAggregate(call(Top2.class, $("vc")).as("value"))
  22.             .flatAggregate(call(Top22.class, $("vc")).as("v", "rank"))
  23.             .select($("id"), $("v"), $("rank"))
  24.             .execute()
  25.             .print();
  26.     }

SQLAPI

  1. // 单值类型
  2.  tEnv.createTemporaryFunction("top2",Top2.class);
  3.  tEnv.sqlQuery("select id,top2(vc) to2_vc from "+ table + " group by id").execute().print();
  5. // 元祖或pojo目前还不支持！！！！

结果：

元祖类型会自动输出为多列，但是需要指定 as （两个元素）：
image.png