• 19、Flink 的Table API 和 SQL 中的自定义函数及示例（2）

    Flink 系列文章

    一、Flink 专栏

    Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

    • 1、Flink 部署系列
      本部分介绍Flink的部署、配置相关基础内容。

    • 2、Flink基础系列
      本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。

    • 3、Flik Table API和SQL基础系列
      本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。

    • 4、Flik Table API和SQL提高与应用系列
      本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。

    • 5、Flink 监控系列
      本部分和实际的运维、监控工作相关。

    二、Flink 示例专栏

    Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

    两专栏的所有文章入口点击：Flink 系列文章汇总索引

    文章目录

    本文介绍了flink的自定义函数概述、开发指南以及标量函数、表值函数的自定义函数实现及说明，提供的示例均可运行并提供运行结果供参考。
    本文依赖flink集群能正常使用。
    本文分为4个部分，即自定义函数的概述、开发指南、标量自定义函数的说明及示例、表值自定义函数的说明及示例。
    本文的示例均在Flink 1.17版本中运行。

    三、自定义函数

    自定义函数（UDF）是一种扩展开发机制，可以用来在查询语句里调用难以用其他方式表达的频繁使用或自定义的逻辑。

    自定义函数可以用 JVM 语言（例如 Java 或 Scala）或 Python 实现，实现者可以在 UDF 中使用任意第三方库，本文聚焦于使用 JVM 语言开发自定义函数。

    1、概述

    当前 Flink 有如下几种函数：

    • 标量函数，将标量值转换成一个新标量值；
    • 表值函数，将标量值转换成新的行数据；
    • 聚合函数，将多行数据里的标量值转换成一个新标量值；
    • 表值聚合函数，将多行数据里的标量值转换成新的行数据；
    • 异步表值函数，是异步查询外部数据系统的特殊函数。

    标量和表值函数已经使用了新的基于数据类型的类型系统，聚合函数仍然使用基于 TypeInformation 的旧类型系统。

    2、开发指南

    在聚合函数使用新的类型系统前，本节仅适用于标量和表值函数。

    所有的自定义函数都遵循一些基本的实现原则。

    1）、函数类

    实现类必须继承自合适的基类之一（例如 org.apache.flink.table.functions.ScalarFunction ）。

    该类必须声明为 public ，而不是 abstract ，并且可以被全局访问。不允许使用非静态内部类或匿名类。

    为了将自定义函数存储在持久化的 catalog 中，该类必须具有默认构造器，且在运行时可实例化。

    Table API 中的匿名函数只有在函数不是有状态的（stateful）（即仅包含瞬态和静态（transient and static）字段）时才能持久化。

    2）、求值方法

    基类提供了一组可以被重写的方法，例如 open()、 close() 或 isDeterministic() 。

    但是，除了上述方法之外，作用于每条传入记录的主要逻辑还必须通过专门的 求值方法 来实现。

    根据函数的种类，后台生成的运算符会在运行时调用诸如 eval()、accumulate() 或 retract() 之类的求值方法。

    这些方法必须声明为 public ，并带有一组定义明确的参数。

    常规的 JVM 方法调用语义是适用的。因此可以：

    • 实现重载的方法，例如 eval(Integer) 和 eval(LocalDateTime)；
    • 使用变长参数，例如 eval(Integer…);
    • 使用对象继承，例如 eval(Object) 可接受 LocalDateTime 和 Integer 作为参数；
    • 也可组合使用，例如 eval(Object…) 可接受所有类型的参数。

    示例片段

    import org.apache.flink.table.functions.ScalarFunction;

// 有多个重载求值方法的函数
public static class SumFunction extends ScalarFunction {

	//两Integer数求和
  public Integer eval(Integer a, Integer b) {
    return a + b;
  }
	//两String数转换后求和
  public Integer eval(String a, String b) {
    return Integer.valueOf(a) + Integer.valueOf(b);
  }

	//多Double数据求和
  public Integer eval(Double... d) {
    double result = 0;
    for (double value : d)
      result += value;
    return (int) result;
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    3）、类型推导

    Table（类似于 SQL 标准）是一种强类型的 API。因此，函数的参数和返回类型都必须映射到数据类型。

    从逻辑角度看，Planner 需要知道数据类型、精度和小数位数；从 JVM 角度来看，Planner 在调用自定义函数时需要知道如何将内部数据结构表示为 JVM 对象。

    术语 类型推导 概括了意在验证输入值、派生出参数/返回值数据类型的逻辑。

    Flink 自定义函数实现了自动的类型推导提取，通过反射从函数的类及其求值方法中派生数据类型。如果这种隐式的反射提取方法不成功，则可以通过使用 @DataTypeHint 和 @FunctionHint 注解相关参数、类或方法来支持提取过程，下面展示了有关如何注解函数的例子。

    如果需要更高级的类型推导逻辑，实现者可以在每个自定义函数中显式重写 getTypeInference() 方法。但是，建议使用注解方式，因为它可使自定义类型推导逻辑保持在受影响位置附近，而在其他位置则保持默认状态。

    1、自动类型推导

    自动类型推导会检查函数的类和求值方法，派生出函数参数和结果的数据类型， @DataTypeHint 和 @FunctionHint 注解支持自动类型推导。

    有关可以隐式映射到数据类型的类的完整列表，请参阅数据类型。

    • @DataTypeHint

    在许多情况下，需要支持以 内联 方式自动提取出函数参数、返回值的类型。

    以下例子展示了如何使用 @DataTypeHint，详情可参考该注解类的文档。

    import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

// 有多个重载求值方法的函数
public static class OverloadedFunction extends ScalarFunction {

  // no hint required
  public Long eval(long a, long b) {
    return a + b;
  }

  // 定义 decimal 的精度和小数位
  public @DataTypeHint("DECIMAL(12, 3)") BigDecimal eval(double a, double b) {
    return BigDecimal.valueOf(a + b);
  }

  // 定义嵌套数据类型
  @DataTypeHint("ROW")
  public Row eval(int i) {
    return Row.of(String.valueOf(i), Instant.ofEpochSecond(i));
  }

  // 允许任意类型的符入，并输出序列化定制后的值
  @DataTypeHint(value = "RAW", bridgedTo = ByteBuffer.class)
  public ByteBuffer eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
    return MyUtils.serializeToByteBuffer(o);
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • @FunctionHint

    有时我们希望一种求值方法可以同时处理多种数据类型，有时又要求对重载的多个求值方法仅声明一次通用的结果类型。

    @FunctionHint 注解可以提供从入参数据类型到结果数据类型的映射，它可以在整个函数类或求值方法上注解输入、累加器和结果的数据类型。可以在类顶部声明一个或多个注解，也可以为类的所有求值方法分别声明一个或多个注解。所有的 hint 参数都是可选的，如果未定义参数，则使用默认的基于反射的类型提取。在函数类顶部定义的 hint 参数被所有求值方法继承。

    以下例子展示了如何使用 @FunctionHint，详情可参考该注解类的文档。

    import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

// 为函数类的所有求值方法指定同一个输出类型
@FunctionHint(output = @DataTypeHint("ROW"))
public static class OverloadedFunction extends TableFunction<Row> {

  public void eval(int a, int b) {
    collect(Row.of("Sum", a + b));
  }

  // overloading of arguments is still possible
  public void eval() {
    collect(Row.of("Empty args", -1));
  }
}

// 解耦类型推导与求值方法，类型推导完全取决于 FunctionHint
@FunctionHint(
  input = {@DataTypeHint("INT"), @DataTypeHint("INT")},
  output = @DataTypeHint("INT")
)
@FunctionHint(
  input = {@DataTypeHint("BIGINT"), @DataTypeHint("BIGINT")},
  output = @DataTypeHint("BIGINT")
)
@FunctionHint(
  input = {},
  output = @DataTypeHint("BOOLEAN")
)
public static class OverloadedFunction extends TableFunction<Object> {

  // an implementer just needs to make sure that a method exists that can be called by the JVM
  public void eval(Object... o) {
    if (o.length == 0) {
      collect(false);
    }
    collect(o[0]);
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    2、定制类型推导

    在大多数情况下，@DataTypeHint 和 @FunctionHint 足以构建自定义函数，然而通过重写 getTypeInference() 定制自动类型推导逻辑，实现者可以创建任意像系统内置函数那样有用的函数。

    以下用 Java 实现的例子展示了定制类型推导的潜力，它根据字符串参数来确定函数的结果类型。该函数带有两个字符串参数：第一个参数表示要分析的字符串，第二个参数表示目标类型。

    import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.types.Row;

public static class LiteralFunction extends ScalarFunction {
  public Object eval(String s, String type) {
    switch (type) {
      case "INT":
        return Integer.valueOf(s);
      case "DOUBLE":
        return Double.valueOf(s);
      case "STRING":
      default:
        return s;
    }
  }

  // 禁用自动的反射式类型推导，使用如下逻辑进行类型推导
  @Override
  public TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()
      // 指定输入参数的类型，必要时参数会被隐式转换
      .typedArguments(DataTypes.STRING(), DataTypes.STRING())
      // specify a strategy for the result data type of the function
      .outputTypeStrategy(callContext -> {
        if (!callContext.isArgumentLiteral(1) || callContext.isArgumentNull(1)) {
          throw callContext.newValidationError("Literal expected for second argument.");
        }
        // 基于字符串值返回数据类型
        final String literal = callContext.getArgumentValue(1, String.class).orElse("STRING");
        switch (literal) {
          case "INT":
            return Optional.of(DataTypes.INT().notNull());
          case "DOUBLE":
            return Optional.of(DataTypes.DOUBLE().notNull());
          case "STRING":
          default:
            return Optional.of(DataTypes.STRING());
        }
      })
      .build();
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45

    4）、确定性

    每个用户自定义函数类都可以通过重写 isDeterministic() 方法来声明它是否产生确定性的结果。如果该函数不是纯粹函数式的（如random(), date(), 或now()），该方法必须返回 false。默认情况下，isDeterministic() 返回 true。

    此外，重写 isDeterministic() 方法也可能影响运行时行为。运行时实现可能会在两个不同的阶段被调用：

    • 在生成执行计划期间：如果一个函数是通过常量表达式调用的或者常量表达式可以从给定的语句中推导出来，那么一个函数就会被预计算以减少常量表达式，并且可能不再在集群上执行。 除非 isDeterministic() 被重写为 false 用来在这种情况下禁用常量表达式简化。比如说，以下对 ABS 的调用在生成执行计划期间被执行：SELECT ABS(-1) FROM t 和 SELECT ABS(field) FROM t WHERE field = -1，而 SELECT ABS(field) FROM t 则不执行。

    • 在运行时（即在集群执行）：如果一个函数被调用时带有非常量表达式或 isDeterministic() 返回 false。

    1、内置函数的确定性

    系统（内置）函数的确定性是不可改变的。存在两种不具有确定性的函数：动态函数和非确定性函数，根据 Apache Calcite SqlOperator 的定义：

      /**
   * Returns whether a call to this operator is guaranteed to always return
   * the same result given the same operands; true is assumed by default.
   */
  public boolean isDeterministic() {
    return true;
  }

  /**
   * Returns whether it is unsafe to cache query plans referencing this
   * operator; false is assumed by default.
   */
  public boolean isDynamicFunction() {
    return false;
  }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    isDeterministic 表示函数的确定性，声明返回 false 时将在运行时对每个记录进行计算。
    isDynamicFunction 声明返回 true 时意味着该函数只能在查询开始时被计算，对于批处理模式，它只在生成执行计划期间被执行， 而对于流模式，它等效于一个非确定性的函数，这是因为查询在逻辑上是连续执行的（流模式对动态表的连续查询抽象），所以动态函数在每次查询执行时也会被重新计算（当前实现下等效于每条记录计算）。

    以下内置函数总是非确定性的（批和流模式下，都在运行时对每条记录进行计算）

    • UUID
    • RAND
    • RAND_INTEGER
    • CURRENT_DATABASE
    • UNIX_TIMESTAMP
    • CURRENT_ROW_TIMESTAMP

    以下内置时间函数是动态的，批处理模式下，将在生成执行计划期间被执行（查询开始），对于流模式，将在运行时对每条记录进行计算

    • CURRENT_DATE
    • CURRENT_TIME
    • CURRENT_TIMESTAMP
    • NOW
    • LOCALTIME
    • LOCALTIMESTAMP

    isDynamicFunction 仅适用于内置函数

    5）、运行时集成

    有时候自定义函数需要获取一些全局信息，或者在真正被调用之前做一些配置（setup）/清理（clean-up）的工作。自定义函数也提供了 open() 和 close() 方法，你可以重写这两个方法做到类似于 DataStream API 中 RichFunction 的功能。

    open() 方法在求值方法被调用之前先调用。close() 方法在求值方法调用完之后被调用。

    open() 方法提供了一个 FunctionContext，它包含了一些自定义函数被执行时的上下文信息，比如 metric group、分布式文件缓存，或者是全局的作业参数等。

    下面的信息可以通过调用 FunctionContext 的对应的方法来获得：

    方法描述
    getMetricGroup()执行该函数的 subtask 的 Metric Group。
    getCachedFile(name)分布式文件缓存的本地临时文件副本。
    getJobParameter(name, defaultValue)跟对应的 key 关联的全局参数值。

    下面的例子展示了如何在一个标量函数中通过 FunctionContext 来获取一个全局的任务参数：

    import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;

public static class HashCodeFunction extends ScalarFunction {

    private int factor = 0;

    @Override
    public void open(FunctionContext context) throws Exception {
        // 获取参数 "hashcode_factor"
        // 如果不存在，则使用默认值 "12"
        factor = Integer.parseInt(context.getJobParameter("hashcode_factor", "12"));
    }

    public int eval(String s) {
        return s.hashCode() * factor;
    }
}

TableEnvironment env = TableEnvironment.create(...);

// 设置任务参数
env.getConfig().addJobParameter("hashcode_factor", "31");

// 注册函数
env.createTemporarySystemFunction("hashCode", HashCodeFunction.class);

// 调用函数
env.sqlQuery("SELECT myField, hashCode(myField) FROM MyTable");

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    3、标量函数-自定义函数说明及示例

    自定义标量函数可以把 0 到多个标量值映射成 1 个标量值，数据类型里列出的任何数据类型都可作为求值方法的参数和返回值类型。

    想要实现自定义标量函数，你需要扩展 org.apache.flink.table.functions 里面的 ScalarFunction 并且实现一个或者多个求值方法。标量函数的行为取决于你写的求值方法。

    求值方法必须是 public 的，而且名字必须是 eval。

    下面自定义函数是将balance加上（万元）以及求balance/age，仅仅示例如何使用，其运行结果在每次输出的代码后面注释的行。

    import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestUDScalarFunctionDemo {

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private int age;
		private int balance;
		private Long rowtime;
	}

	final static List<User> userList = Arrays.asList(
			new User(1L, "alan", 18, 20,1698742358391L), 
			new User(2L, "alan", 19, 25,1698742359396L), 
			new User(3L, "alan", 25, 30,1698742360407L),
			new User(4L, "alanchan", 28,35, 1698742361409L), 
			new User(5L, "alanchan", 29, 35,1698742362424L)
			);

	public static class TestScalarFunction extends ScalarFunction {

		// 接受任意类型输入，返回 String 型输出
		public String eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
			return o.toString() + " (万元)";
		}

		public double eval(Integer  age, Integer  balance) {
			return balance / age *1.0;
		}

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		DataStream<User> users = env.fromCollection(userList);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("age"),$("balance"), $("rowtime"));

		//1、 在 Table API 里不经注册直接“内联”调用函数
		Table result = usersTable.select($("id"), $("name"), call(TestScalarFunction.class, $("balance")));

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
//		resultDS.print();
//		11> (true,+I[2, alan, 25 (万元)])
//		12> (true,+I[3, alan, 30 (万元)])
//		13> (true,+I[4, alanchan, 35 (万元)])
//		10> (true,+I[1, alan, 20 (万元)])
//		14> (true,+I[5, alanchan, 35 (万元)])

		Table result2 = usersTable.select($("id"), $("name"), $("age"), 
				call(TestScalarFunction.class, $("balance")), 
				call(TestScalarFunction.class, $("age"), $("balance"))
				);
		
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
//		result2DS.print();
//		9> (true,+I[2, alan, 19, 25 (万元), 1.0])
//		10> (true,+I[3, alan, 25, 30 (万元), 1.0])
//		12> (true,+I[5, alanchan, 29, 35 (万元), 1.0])
//		11> (true,+I[4, alanchan, 28, 35 (万元), 1.0])
//		8> (true,+I[1, alan, 18, 20 (万元), 1.0])
		
		//2、 注册函数
		tenv.createTemporarySystemFunction("TestScalarFunction", TestScalarFunction.class);
		// 在 Table API 里调用注册好的函数
		Table result3 = usersTable.select($("id"), $("name"),call("TestScalarFunction", $("balance")));
		
		DataStream<Tuple2<Boolean, Row>> result3DS = tenv.toRetractStream(result3, Row.class);
//		result3DS.print();
//		2> (true,+I[4, alanchan, 35 (万元)])
//		3> (true,+I[5, alanchan, 35 (万元)])
//		15> (true,+I[1, alan, 20 (万元)])
//		16> (true,+I[2, alan, 25 (万元)])
//		1> (true,+I[3, alan, 30 (万元)])
		
		// 在 SQL 里调用注册好的函数
		 tenv.createTemporaryView("user_view", users);
		
		Table result4 = tenv.sqlQuery("SELECT id,name,TestScalarFunction(balance) ,TestScalarFunction(age,balance) FROM user_view");
		
		DataStream<Tuple2<Boolean, Row>> result4DS = tenv.toRetractStream(result4, Row.class);
		result4DS.print();
//		14> (true,+I[1, alan, 20 (万元), 1.0])
//		1> (true,+I[4, alanchan, 35 (万元), 1.0])
//		2> (true,+I[5, alanchan, 35 (万元), 1.0])
//		15> (true,+I[2, alan, 25 (万元), 1.0])
//		16> (true,+I[3, alan, 30 (万元), 1.0])
		
		env.execute();
	}

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123

    4、表值函数-自定义函数说明及示例

    跟自定义标量函数一样，自定义表值函数的输入参数也可以是 0 到多个标量。但是跟标量函数只能返回一个值不同的是，它可以返回任意多行。返回的每一行可以包含 1 到多列，如果输出行只包含 1 列，会省略结构化信息并生成标量值，这个标量值在运行阶段会隐式地包装进行里。

    要定义一个表值函数，你需要扩展 org.apache.flink.table.functions 下的 TableFunction，可以通过实现多个名为 eval 的方法对求值方法进行重载。像其他函数一样，输入和输出类型也可以通过反射自动提取出来。表值函数返回的表的类型取决于 TableFunction 类的泛型参数 T，不同于标量函数，表值函数的求值方法本身不包含返回类型，而是通过 collect(T) 方法来发送要输出的行。

    在 Table API 中，表值函数是通过 .joinLateral(…) 或者 .leftOuterJoinLateral(…) 来使用的。joinLateral 算子会把外表（算子左侧的表）的每一行跟跟表值函数返回的所有行（位于算子右侧）进行 （cross）join。leftOuterJoinLateral 算子也是把外表（算子左侧的表）的每一行跟表值函数返回的所有行（位于算子右侧）进行（cross）join，并且如果表值函数返回 0 行也会保留外表的这一行。

    在 SQL 里面用 JOIN 或者 以 ON TRUE 为条件的 LEFT JOIN 来配合 LATERAL TABLE() 的使用。

    下面示例中包含表值函数的四种应用方式。

    import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestUDTableFunctionDemo {

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private int age;
		private int balance;
		private Long rowtime;
	}

	final static List<User> userList = Arrays.asList(
			new User(1L, "alan,chen", 18, 20,1698742358391L), 
			new User(2L, "alan,chen", 19, 25,1698742359396L), 
			new User(3L, "alan,chen", 25, 30,1698742360407L),
			new User(4L, "alan,chan", 28,35, 1698742361409L), 
			new User(5L, "alan,chan", 29, 35,1698742362424L)
			);
	
	@FunctionHint(output = @DataTypeHint("ROW"))
	public static class SplitFunction extends TableFunction<Row> {

		public void eval(String str) {
			String[] names = str.split(",");
			collect(Row.of(names[0],names[1]));
//			for (String s : str.split(", ")) {
//				// use collect(...) to emit a row
//				collect(Row.of(s, s.length()));
//			}
		}
	}
	
	@FunctionHint(output = @DataTypeHint("ROW"))
	public static class OverloadedFunction extends TableFunction<Row> {

		public void eval(String str) {
			String[] user = str.split(",");
			collect(Row.of(Integer.valueOf(user[0]),user[1],Integer.valueOf(user[2]),Integer.valueOf(user[3]),user[4]));
		}
	}

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		DataStream<User> users = env.fromCollection(userList);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("age"), $("balance"), $("rowtime"));

		// 1、 在 Table API 里不经注册直接“内联”调用函数
		Table result = usersTable
				 .joinLateral(call(SplitFunction.class, $("name")))
				.select($("id"), $("name"),$("firstName"),$("lastName"));

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
//		resultDS.print();
//		11> (true,+I[5, alan,chan, alan, chan])
//		7> (true,+I[1, alan,chen, alan, chen])
//		9> (true,+I[3, alan,chen, alan, chen])
//		10> (true,+I[4, alan,chan, alan, chan])
//		8> (true,+I[2, alan,chen, alan, chen])
		
		DataStream<String> row = env.fromCollection(
				//id name age balance rowtime
				Arrays.asList(
						"11,alan,18,20,1699341167461",
						"12,alan,19,25,1699341168464",
						"13,alan,20,30,1699341169472",
						"14,alanchan,18,22,1699341170479",
						"15,alanchan,19,25,1699341171482"
						)
				);
		Table usersTable2 = tenv.fromDataStream(row, $("userString"));
		
		Table result2 = usersTable2
							 .joinLateral(call(OverloadedFunction.class, $("userString")))
							 .select($("userString"),$("id"),$("name"),$("age"),$("balance"),$("rowtime")
						 )	;	
						
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
//		result2DS.print();
//		15> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//		13> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//		14> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//		11> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//		12> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
		
		Table result3 = usersTable2
				 .leftOuterJoinLateral(call(OverloadedFunction.class, $("userString")))
				 .select($("userString"),$("id"),$("name"),$("age"),$("balance"),$("rowtime")
			 )	;	
			
		DataStream<Tuple2<Boolean, Row>> result3DS = tenv.toRetractStream(result3, Row.class);
//		result3DS.print();
//		5> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//		6> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//		3> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//		4> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
//		7> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
		
		// 在 Table API 里重命名函数字段
		Table result4 = usersTable2
				 .leftOuterJoinLateral(call(OverloadedFunction.class, $("userString")).as("t_id","t_name","t_age","t_balance","t_rowtime"))
				 .select($("userString"),$("t_id"),$("t_name"),$("t_age"),$("t_balance"),$("t_rowtime")
			 )	;	
			
		DataStream<Tuple2<Boolean, Row>> result4DS = tenv.toRetractStream(result4, Row.class);
//		result4DS.print();
//		10> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//		13> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//		14> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//		12> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//		11> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
		
		//2、 注册函数
		tenv.createTemporarySystemFunction("OverloadedFunction", OverloadedFunction.class);
		
		// 在 Table API 里调用注册好的函数
		Table result5 = usersTable2
				 .leftOuterJoinLateral(call("OverloadedFunction", $("userString")).as("t_id","t_name","t_age","t_balance","t_rowtime"))
				 .select($("userString"),$("t_id"),$("t_name"),$("t_age"),$("t_balance"),$("t_rowtime")
			 )	;	
			
		DataStream<Tuple2<Boolean, Row>> result5DS = tenv.toRetractStream(result5, Row.class);
//		result5DS.print();
//		11> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//		14> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//		15> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//		13> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//		12> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
		
		Table result6 = usersTable2
				 .joinLateral(call("OverloadedFunction", $("userString")).as("t_id","t_name","t_age","t_balance","t_rowtime"))
				 .select($("userString"),$("t_id"),$("t_name"),$("t_age"),$("t_balance"),$("t_rowtime")
			 )	;	
			
		DataStream<Tuple2<Boolean, Row>> result6DS = tenv.toRetractStream(result6, Row.class);
//		result6DS.print();
//		8> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//		9> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//		5> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//		7> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//		6> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
		
		//3、 在 SQL 里调用注册好的函数
		 tenv.createTemporaryView("user_view", usersTable2);
		 Table result7 =  tenv.sqlQuery(
				  "SELECT userString, id,name,age,balance,rowtime " +
				  "FROM user_view, LATERAL TABLE(OverloadedFunction(userString))");
			DataStream<Tuple2<Boolean, Row>> result7DS = tenv.toRetractStream(result7, Row.class);
//			result7DS.print();
//			15> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//			13> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//			1> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//			14> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
//			16> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
		
			 Table result8 =  tenv.sqlQuery(
					  "SELECT userString, id,name,age,balance,rowtime " +
					  "FROM user_view "+
					  " LEFT JOIN LATERAL TABLE( OverloadedFunction(userString)) ON TRUE  "  );
				DataStream<Tuple2<Boolean, Row>> result8DS = tenv.toRetractStream(result8, Row.class);
//				result8DS.print();
//				13> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
//				1> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//				15> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//				14> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
//				16> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
			
				//4、 在 SQL 里重命名函数字段
				 Table result9 =  tenv.sqlQuery(
						  "SELECT userString, t_id, t_name,t_age,t_balance,t_rowtime " +
						  "FROM user_view "+
						 "LEFT JOIN LATERAL TABLE(OverloadedFunction(userString)) AS T(t_id, t_name,t_age,t_balance,t_rowtime) ON TRUE");
					DataStream<Tuple2<Boolean, Row>> result9DS = tenv.toRetractStream(result9, Row.class);
					result9DS.print();
//					7> (true,+I[12,alan,19,25,1699341168464, 12, alan, 19, 25, 1699341168464])
//					10> (true,+I[15,alanchan,19,25,1699341171482, 15, alanchan, 19, 25, 1699341171482])
//					9> (true,+I[14,alanchan,18,22,1699341170479, 14, alanchan, 18, 22, 1699341170479])
//					8> (true,+I[13,alan,20,30,1699341169472, 13, alan, 20, 30, 1699341169472])
//					6> (true,+I[11,alan,18,20,1699341167461, 11, alan, 18, 20, 1699341167461])
					
		env.execute();
		
	}

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158
    • 159
    • 160
    • 161
    • 162
    • 163
    • 164
    • 165
    • 166
    • 167
    • 168
    • 169
    • 170
    • 171
    • 172
    • 173
    • 174
    • 175
    • 176
    • 177
    • 178
    • 179
    • 180
    • 181
    • 182
    • 183
    • 184
    • 185
    • 186
    • 187
    • 188
    • 189
    • 190
    • 191
    • 192
    • 193
    • 194
    • 195
    • 196
    • 197
    • 198
    • 199
    • 200
    • 201
    • 202
    • 203
    • 204
    • 205
    • 206
    • 207
    • 208
    • 209
    • 210
    • 211
    • 212
    • 213
    • 214
    • 215
    • 216

    以上，介绍了flink的自定义函数概述、开发指南以及标量函数、表值函数的自定义函数实现及说明，提供的示例均可运行并提供运行结果供参考。

  • 相关阅读:
    【第四篇】SpringSecurity的HttpSecurity详解
    【经验模态分解】2.EMD的3个基本概念
    类型体系与基本数据类型（第三节）
    计算机网络--物理层练习题
    前沿研究|16s+宏基因组binning揭示大型藻类附生微生物群落核心组成
    An工具介绍之形状工具及渐变变形工具
    【Python数据分析 - 5】：Numpy-数组的基本操作
    Java日志
    纯血鸿蒙开发教程：如何实现运动饮食卡片效果
    jenkins持续集成环境搭建
  • 原文地址：https://blog.csdn.net/chenwewi520feng/article/details/132731154