背景

flink本身給我們提供了大量的內置函數，已經能滿足我們絕大部分的需求，但是如果確實是碰到了一些特殊的場景，無法滿足我們的需求的時候，我們可以自定義函數來解決，今天我們主要講一下最簡單的自定義函數-ScalarFunction。

實例講解

定義函數

首先我們來實現一個最簡單的標量函數，主要功能就是對兩個數加和操作，實現標量函數需要繼承抽象類ScalarFunction，然后實現eval方法，代碼如下：

 public static class SumFunction extends ScalarFunction{

  public Integer eval(Integer a, Integer b){
   return a + b;
  }
 }

通過程序注冊函數

 //通過程序的方式來注冊函數
  SumFunction sumFunction = new SumFunction();
  tableEnv.registerFunction("mysum", sumFunction);
  Table table1 = tableEnv.sqlQuery("select mysum(1,2)");
  tableEnv.toAppendStream(table1, Row.class).print();

通過sql注冊函數

 //通過sql的方式來注冊函數
  String className = SumFunction.class.getName();
  String sql = "create temporary function default_catalog.default_database.mysum1" +
               " as '" + className + "'";
  tableEnv.sqlUpdate(sql);
  Table table2 = tableEnv.sqlQuery("select mysum1(3,4)");
  tableEnv.toAppendStream(table2, Row.class).print();

之后我們可以通過StreamTableEnvironment類的listFunctions方法列出來所有的函數來看下我們自定義的函數是否在其中.

 //列出來所有的函數，看是否包含我們定義的函數
  String[] functions = tableEnv.listFunctions();
  Stream.of(functions).filter(f->f.startsWith("mysum")).forEach(System.out::println);

最后如果自定義的jar和主程序不在一個jar包，通過命令行提交任務的時候，記得通過參數--classpath（簡寫-C）將包含自定義函數的jar添加到classpath里。