Flink自定義函數(shù)之標(biāo)量函數(shù)怎么用

背景

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

實(shí)例講解

定義函數(shù)

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

 public static class SumFunction extends ScalarFunction{

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

通過程序注冊函數(shù)

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

通過sql注冊函數(shù)

 //通過sql的方式來注冊函數(shù)
  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方法列出來所有的函數(shù)來看下我們自定義的函數(shù)是否在其中.

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

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