标量函数ScalarFunction：一对一

  • 输入一行 输出一行 ```java // 注册函数 // createTemporarySystemFunction : 在全局注册 // createTemporaryFunction : 在当前catalog+database注册 tableEnv.createTemporarySystemFunction(“MyHash”, MyHash.class);

//TableAPI tableEnv.from(“MyTable”) .select($(“user”), call(“MyHash”, $(“myField”))); //SQL tableEnv.sqlQuery(“SELECT MyHash(myField) FROM MyTable”);

// 实现自定义的ScalarFunction public static class MyHash extends ScalarFunction { // 实现eval方法 // @DataTypeHint(inputGroup = InputGroup.ANY) 表示任何类型 public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o){ return o.hashCode(); } }

  1. <a name="PV10P"></a>
  2. ### 表函数TableFunction：一对多
  4. - 输入一行 输出一个或多个
  6. 输入：sensor_6,123,123 <br />输出：<br />sensor_6,123,123,sensor,6<br />sensor_6,123,123,6,1
  7. ```java
  8. // 需要在环境中注册UDF
  9. tableEnv.createTemporarySystemFunction("MySplit", MySplit.class);
  11. //TableAPI
  12. Table resultTable = sensorTable
  13.         .joinLateral("split(id) as (word, length)")
  14.         .select("id, ts, word, length");
  15. //SQL
  16. select 
  17. user, 
  18. url, 
  19. word, 
  20. length
  21. from EventTable, 
  22. LATERAL TABLE( MySplit(url) ) AS T(word, length)
  24. // 实现自定义TableFunction
  25. // 注意有泛型，这里输出的是两个字段，二元组
  26. // DataTypeHint: 为TableAPI使用而手动声明的类型
  27. @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
  28. public static class MySplit extends TableFunction<Tuple2<String, Integer>>{ //泛型是输出类型
  29.     // 必须实现一个eval方法，没有返回值，按照？分割
  30.     public void eval( String str ){
  31.         for( String s: str.split("\\?") ){
  32.             collect(new Tuple2<>(s, s.length()));
  33.         }
  34.     }
  35. }

聚合函数AggregateFunction：多对一

  • 输入多行
  • AggregationFunction要求必须实现的方法:
    • createAccumulator()
    • accumulate() :与之前的 eval()类似，也是底层架构要求的，必须为 public， 方法名必须为 accumulate，且无法直接 override、只能手动实现
    • getValue()
  • AggregateFunction 的工作原理如下:
    • 首先，它需要一个累加器(Accumulator)，用来保存聚合中间结果的数据结构;可以通过调用 createAccumulator() 方法创建空累加器
    • 随后，对每个输入行调用函数的 accumulate() 方法来更新累加器
    • 处理完所有行后，将调用函数的 getValue() 方法来计算并返回最终结果 ```java //TableAPI Table resultTable = sensorTable .groupBy(“id”) .aggregate(“avgTemp(temp) as avgtemp” ) .select(“id, avgtemp”); //SQL tableEnv.createTemporaryView(“sensor”, sensorTable); Table resultSqlTable = tableEnv.sqlQuery(
      1.    "select " +
      2.    "id, " +
      3.    "avgTemp(temp) " +
      4.    "from sensor " +
      5.    "group by id");

// 实现自定义的AggregateFunction //> : <输出类型，累加器类型> public static class AvgTemp extends AggregateFunction>{ @Override public Tuple2 createAccumulator() { return new Tuple2<>(0.0, 0);// 温度总和,个数 }

  1. public void accumulate( Tuple2<Double, Integer> accumulator, Double temp ){
  2.     accumulator.f0 += temp;
  3.     accumulator.f1 += 1;
  4. }
  6. @Override
  7. public Double getValue(Tuple2<Double, Integer> accumulator) {
  8.     return accumulator.f0 / accumulator.f1;
  9. }

}

  1. <a name="xnbUu"></a>
  2. ### 表聚合函数TableAggregateFunction：多对多
  4. - 输入多行 输出多行，典型应用场景TOPN
  5.    - 输入：sensor_1,1547718199,35.8<br />            sensor_1,1547718207,36.3<br />            sensor_1,1547718212,37.1
  6.    - 输出：1> (true,sensor_1,35.8,1)
  8. 1> (true,sensor_1,4.9E-324,2)<br />1> (false,sensor_1,35.8,1)<br />1> (false,sensor_1,4.9E-324,2)<br />1> (true,sensor_1,37.1,1)<br />1> (true,sensor_1,35.8,2)<br />1> (false,sensor_1,37.1,1)<br />1> (false,sensor_1,35.8,2)<br />1> (true,sensor_1,37.1,1)<br />1> (true,sensor_1,36.3,2)
  10. - AggregationFunction 要求必须实现的方法: 
  11.    -  createAccumulator()
  12.    - accumulate() 
  13.    - emitValue() ：无法override 手动实现
  14. - TableAggregateFunction 的工作原理如下:
  15.    - 首先，它同样需要一个累加器(Accumulator)，它是保存聚合中间结果的数据结构。通过调用 createAccumulator() 方法可以创建空累加器。
  16.    - 随后，对每个输入行调用函数的 accumulate() 方法来更新累加器。
  17.    - 处理完所有行后，将调用函数的 emitValue() 方法来计算并返回最终结果。
  18. ```java
  19. // 创建一个表聚合函数实例
  20. Top2Temp top2Temp = new Top2Temp();
  21. tableEnv.registerFunction("top2Temp", top2Temp);
  23. // table api使用
  24. Table resultTable = sensorTable
  25.     .groupBy("id")
  26.     .flatAggregate("top2Temp(temperature) as (temp, rank)")
  27.     .select("id, temp, rank");
  28. // SQL中没有表聚合函数的使用方式
  32. // 表聚合函数需要使用的累加器
  33. public static class Top2TempAcc {
  34.     double highestTemp = Double.MIN_VALUE;
  35.     double secondHighestTemp = Double.MIN_VALUE;
  36. }
  37. // 自定义表聚合函数
  38. public static class Top2Temp extends TableAggregateFunction<Tuple2<Double,Integer>, Top2TempAcc> {
  39.     @Override
  40.     public Top2TempAcc createAccumulator() {
  41.         return new Top2TempAcc();
  42.     }
  44.     public void accumulate(Top2TempAcc acc, Double temp) {
  45.         if (temp > acc.highestTemp) {
  46.             acc.secondHighestTemp = acc.highestTemp;
  47.             acc.highestTemp = temp;
  48.         } else if (temp > acc.secondHighestTemp) {
  49.             acc.secondHighestTemp = temp;
  50.         }
  51.     }
  53.     public void emitValue(Top2TempAcc acc, Collector<Tuple2<Double, Integer>> out) {
  54.         out.collect(new Tuple2<>(acc.highestTemp, 1));
  55.         out.collect(new Tuple2<>(acc.secondHighestTemp, 2));
  56.     }
  57. }