配置管理UDF

前提条件

准备UDF jar文件，大小不能超过200MB。

上传UDF

1. 访问Flink WebUI，请参考访问Flink WebUI。
2. 单击“UDF管理”进入UDF管理页面。
3. 单击“添加UDF”，在“本地Jar文件”参数后选择并上传本地已准备好的UDF jar文件。
4. 填写UDF名称以及描述信息后，单击“确定”。
   

   • “UDF名称”最多可添加10项，“名称”可自定义，“类名”需与上传的UDF jar文件中UDF函数全限定类名一一对应。
   • 上传UDF jar文件后，服务器默认保留5分钟，5分钟内单击确定则完成UDF创建，超时后单击确定则创建UDF失败并弹出错误提示：本地UDF文件路径有误。

5. 在UDF列表中，可查看当前应用内所有的UDF信息。可在对应UDF信息的“操作”列编辑或删除UDF信息（只能删除未被使用的UDF项）。
6. （可选）如果需要立即运行或开发作业，可在“作业管理”进行相关作业配置，可参考创建作业。

UDF java代码及SQL样例

• UDF java使用样例

  package com.xxx.udf;
  import org.apache.flink.table.functions.ScalarFunction;
  public class UdfClass_UDF extends ScalarFunction {
      public int eval(String s) {
          return s.length();
      }
  }

• UDF SQL使用样例

  CREATE TEMPORARY FUNCTION udf as 'com.xxx.udf.UdfClass_UDF';
  CREATE TABLE udfSource (a VARCHAR) WITH ('connector' = 'datagen','rows-per-second'='1');
  CREATE TABLE udfSink (a VARCHAR,b int) WITH ('connector' = 'print');
  INSERT INTO
    udfSink
  SELECT
    a,
    udf(a)
  FROM
    udfSource;

UDAF java代码及SQL样例

• UDAF java使用样例

  package com.xxx.udf;
  import org.apache.flink.table.functions.AggregateFunction;
  public class UdfClass_UDAF {
      public static class AverageAccumulator  {
          public int sum;
      }
      public static class Average extends AggregateFunction<Integer, AverageAccumulator> {
          public void accumulate(AverageAccumulator acc, Integer value) {
              acc.sum += value;
          }
          @Override
          public Integer getValue(AverageAccumulator acc) {
              return acc.sum;
          }
          @Override
          public AverageAccumulator createAccumulator() {
              return new AverageAccumulator();
          }
      }
  }

UDTF java代码及SQL样例

• UDTF java使用样例

  package com.xxx.udf;
  import org.apache.flink.api.java.tuple.Tuple2;
  import org.apache.flink.table.functions.TableFunction;
  public class UdfClass_UDTF extends TableFunction<Tuple2<String, Integer>> {
      public void eval(String str) {
          Tuple2<String, Integer> tuple2 = Tuple2.of(str, str.length());
          collect(tuple2);
      }
  }