사용자 정의 함수

사용자 정의 함수 (User-defined Functions)

사용자 정의 함수(User-defined Function, UDF)는 자주 사용하는 로직이나 쿼리로는 표현할 수 없는 사용자 지정 로직을 호출하기 위한 확장 지점입니다.

사용자 정의 함수는 JVM 언어(예: Java 또는 Scala)나 Python으로 구현할 수 있습니다. 구현자는 UDF 안에서 임의의 서드파티 라이브러리를 사용할 수 있습니다. 이 페이지는 JVM 기반 언어에 초점을 맞추며, Python에서 일반적인 UDF와 벡터화된 UDF를 작성하는 방법에 대한 자세한 내용은 PyFlink 문서를 참조하세요.

출처: 문서

본문

개요 (Overview)

현재 Flink는 다음과 같은 종류의 함수를 구분합니다.

  • 스칼라 함수(Scalar functions): 스칼라 값을 새로운 스칼라 값으로 매핑합니다.
  • 비동기 스칼라 함수(Asynchronous scalar functions): 스칼라 값을 비동기적으로 새로운 스칼라 값으로 매핑합니다.
  • 테이블 함수(Table functions): 스칼라 값을 새로운 행(row)으로 매핑합니다.
  • 비동기 테이블 함수(Async table functions): 스칼라 값을 비동기적으로 새로운 행으로 매핑하며, 룩업을 수행하는 테이블 소스에 사용할 수 있습니다.
  • 집계 함수(Aggregate functions): 여러 행의 스칼라 값을 새로운 스칼라 값으로 매핑합니다.
  • 테이블 집계 함수(Table aggregate functions): 여러 행의 스칼라 값을 새로운 행으로 매핑합니다.
  • 프로세스 테이블 함수(Process table functions): 테이블을 새로운 행으로 매핑합니다. 상태(state)와 타이머(timer)를 가진 사용자 정의 연산자를 활성화합니다.

다음 예제는 간단한 스칼라 함수를 만들고 Table API와 SQL 양쪽에서 함수를 호출하는 방법을 보여줍니다.

SQL 쿼리에서는 함수를 항상 이름으로 등록해야 합니다. Table API에서는 함수를 등록하거나 인라인으로 직접 사용할 수 있습니다.

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;

// define function logic
public static class SubstringFunction extends ScalarFunction {
  public String eval(String s, Integer begin, Integer end) {
    return s.substring(begin, end);
  }
}

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

// call function "inline" without registration in Table API
env.from("MyTable").select(call(SubstringFunction.class, $("myField"), 5, 12));

// register function
env.createTemporarySystemFunction("SubstringFunction", SubstringFunction.class);

// call registered function in Table API
env.from("MyTable").select(call("SubstringFunction", $("myField"), 5, 12));

// call registered function in SQL
env.sqlQuery("SELECT SubstringFunction(myField, 5, 12) FROM MyTable");

// call registered function in SQL using named parameters
env.sqlQuery("SELECT SubstringFunction(param1 => myField, param2 => 5, param3 => 12) FROM MyTable");

import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction

// define function logic
class SubstringFunction extends ScalarFunction {
  def eval(s: String, begin: Integer, end: Integer): String = {
    s.substring(begin, end)
  }
}

val env = TableEnvironment.create(...)

// call function "inline" without registration in Table API
env.from("MyTable").select(call(classOf[SubstringFunction], $"myField", 5, 12))

// register function
env.createTemporarySystemFunction("SubstringFunction", classOf[SubstringFunction])

// call registered function in Table API
env.from("MyTable").select(call("SubstringFunction", $"myField", 5, 12))

// call registered function in SQL
env.sqlQuery("SELECT SubstringFunction(myField, 5, 12) FROM MyTable")

대화형 세션에서는 함수를 사용하거나 등록하기 전에 파라미터화할 수 있습니다. 이 경우 함수 클래스 대신 함수 인스턴스를 임시 함수로 사용할 수 있습니다.

함수 인스턴스를 클러스터로 보내려면 파라미터가 직렬화 가능해야 합니다.

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;

// define parameterizable function logic
public static class SubstringFunction extends ScalarFunction {

  private boolean endInclusive;

  public SubstringFunction(boolean endInclusive) {
    this.endInclusive = endInclusive;
  }

  public String eval(String s, Integer begin, Integer end) {
    return s.substring(begin, endInclusive ? end + 1 : end);
  }
}

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

// call function "inline" without registration in Table API
env.from("MyTable").select(call(new SubstringFunction(true), $("myField"), 5, 12));

// register function
env.createTemporarySystemFunction("SubstringFunction", new SubstringFunction(true));

import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction

// define parameterizable function logic
class SubstringFunction(val endInclusive) extends ScalarFunction {
  def eval(s: String, begin: Integer, end: Integer): String = {
    s.substring(endInclusive ? end + 1 : end)
  }
}

val env = TableEnvironment.create(...)

// call function "inline" without registration in Table API
env.from("MyTable").select(call(new SubstringFunction(true), $"myField", 5, 12))

// register function
env.createTemporarySystemFunction("SubstringFunction", new SubstringFunction(true))

Table API에서 함수 호출의 인자로 별표 * 표현식을 사용해 와일드카드처럼 동작하게 할 수 있습니다. 그러면 테이블의 모든 컬럼이 해당 위치의 함수로 전달됩니다.

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;

public static class MyConcatFunction extends ScalarFunction {
  public String eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object... fields) {
    return Arrays.stream(fields)
        .map(Object::toString)
        .collect(Collectors.joining(","));
  }
}

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

// call function with $("*"), if MyTable has 3 fields (a, b, c),
// all of them will be passed to MyConcatFunction.
env.from("MyTable").select(call(MyConcatFunction.class, $("*")));

// it's equal to call function with explicitly selecting all columns.
env.from("MyTable").select(call(MyConcatFunction.class, $("a"), $("b"), $("c")));

import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction

import scala.annotation.varargs

class MyConcatFunction extends ScalarFunction {
  @varargs
  def eval(@DataTypeHint(inputGroup = InputGroup.ANY) row: AnyRef*): String = {
    row.map(f => f.toString).mkString(",")
  }
}

val env = TableEnvironment.create(...)

// call function with $"*", if MyTable has 3 fields (a, b, c),
// all of them will be passed to MyConcatFunction.
env.from("MyTable").select(call(classOf[MyConcatFunction], $"*"));

// it's equal to call function with explicitly selecting all columns.
env.from("MyTable").select(call(classOf[MyConcatFunction], $"a", $"b", $"c"));

TableEnvironment는 UserDefinedFunction로 임시 시스템 함수를 만드는 두 가지 오버로드 메서드를 제공합니다.

  • createTemporarySystemFunction(String name, Class<? extends UserDefinedFunction> functionClass)
  • createTemporarySystemFunction(String name, UserDefinedFunction functionInstance)

사용자 정의 함수에 인자가 없는 생성자가 있다면 functionClass를 functionInstance보다 사용하는 것이 권장됩니다. 이는 프레임워크로서의 Flink가 새 인스턴스를 생성하는 과정을 제어하는 더 많은 로직을 추가할 수 있기 때문입니다. TableEnvironmentImpl의 현재 내장 표준 로직은 UserDefinedFunction의 서브클래스 타입(예: ScalarFunction, TableFunction)에 따라 클래스와 클래스의 메서드를 검증합니다. 사용자 기존 코드를 변경할 필요 없이 향후 프레임워크에 더 많은 로직이나 최적화가 추가될 수 있습니다.

구현 가이드 (Implementation Guide)

함수의 종류와 무관하게 모든 사용자 정의 함수는 몇 가지 기본적인 구현 원칙을 따릅니다.

함수 클래스 (Function Class)

구현 클래스는 사용 가능한 기본 클래스 중 하나(예: org.apache.flink.table.functions.ScalarFunction)를 확장해야 합니다.

클래스는 public으로 선언되어야 하고 abstract가 아니어야 하며 전역적으로 접근 가능해야 합니다. 즉 비정적(non-static) 내부 클래스나 익명(anonymous) 클래스는 허용되지 않습니다.

사용자 정의 함수를 영구 카탈로그에 저장하려면 클래스에 기본 생성자가 있어야 하며 런타임에 인스턴스화할 수 있어야 합니다. Table API의 익명 함수는 함수가 상태를 갖지 않는 경우(즉 transient 및 static 필드만 포함)에만 영구 저장될 수 있습니다.

평가 메서드 (Evaluation Methods)

기본 클래스는 open(), close(), isDeterministic() 또는 supportsConstantFolding()과 같이 오버라이드할 수 있는 메서드 집합을 제공합니다.

그러나 선언된 메서드 외에도 들어오는 모든 레코드에 적용되는 주요 런타임 로직은 전문화된 평가 메서드를 통해 구현해야 합니다.

함수 종류에 따라 eval(), accumulate(), retract() 같은 평가 메서드가 런타임에 코드 생성된 연산자에 의해 호출됩니다.

메서드는 public으로 선언되어야 하며 잘 정의된 인자 집합을 가져야 합니다.

일반 JVM 메서드 호출 의미가 적용됩니다. 따라서 다음이 가능합니다.

  • eval(Integer)와 eval(LocalDateTime) 같은 오버로드된 메서드 구현
  • eval(Integer...) 같은 가변 인자(var-args) 사용
  • LocalDateTime과 Integer를 모두 받는 eval(Object) 같은 객체 상속 사용
  • 모든 종류의 인자를 받는 eval(Object...) 같은 위의 조합

Scala에서 함수를 구현하려면 가변 인자일 때 scala.annotation.varargs 어노테이션을 추가하세요. 또한 NULL을 지원하려면 박스형 기본 타입(예: Int 대신 java.lang.Integer)을 사용하는 것이 권장됩니다.

다음 스니펫은 오버로드된 함수의 예를 보여줍니다.

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

// function with overloaded evaluation methods
public static class SumFunction extends ScalarFunction {

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

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

  public Integer eval(Double... d) {
    double result = 0;
    for (double value : d)
      result += value;
    return (int) result;
  }
}

import org.apache.flink.table.functions.ScalarFunction
import java.lang.Integer
import java.lang.Double
import scala.annotation.varargs

// function with overloaded evaluation methods
class SumFunction extends ScalarFunction {

  def eval(a: Integer, b: Integer): Integer = {
    a + b
  }

  def eval(a: String, b: String): Integer = {
    Integer.valueOf(a) + Integer.valueOf(b)
  }

  @varargs // generate var-args like Java
  def eval(d: Double*): Integer = {
    d.sum.toInt
  }
}

타입 추론 (Type Inference)

테이블 생태계(SQL 표준과 유사)는 강력한 타입의 API입니다. 따라서 함수 파라미터와 반환 타입 모두 데이터 타입에 매핑되어야 합니다.

논리적 관점에서 플래너는 예상 타입, 정밀도(precision), 스케일(scale)에 대한 정보가 필요합니다. JVM 관점에서 플래너는 사용자 정의 함수를 호출할 때 내부 데이터 구조가 JVM 객체로 어떻게 표현되는지에 대한 정보가 필요합니다.

입력 인자를 검증하고 함수의 파라미터와 결과에 대한 데이터 타입을 도출하는 로직을 총칭하여 타입 추론(type inference)이라고 합니다.

Flink의 사용자 정의 함수는 리플렉션을 통해 함수의 클래스와 평가 메서드에서 데이터 타입을 도출하는 자동 타입 추론 추출을 구현합니다. 이 암시적 리플렉션 추출 방식이 성공하지 못하면 @DataTypeHint와 @FunctionHint로 영향을 받는 파라미터, 클래스, 메서드에 어노테이션을 달아 추출 과정을 보완할 수 있습니다. 함수에 어노테이션을 다는 더 많은 예는 아래에 나와 있습니다.

더 고급스러운 타입 추론 로직이 필요하면 구현자는 모든 사용자 정의 함수에서 getTypeInference() 메서드를 명시적으로 오버라이드할 수 있습니다. 그러나 커스텀 타입 추론 로직을 영향받는 위치 가까이 유지하고 나머지 구현은 기본 동작에 폴백하므로 어노테이션 방식을 권장합니다.

자동 타입 추론 (Automatic Type Inference)

자동 타입 추론은 함수의 클래스와 평가 메서드를 검사해 함수의 인자와 결과에 대한 데이터 타입을 도출합니다. @DataTypeHint와 @FunctionHint 어노테이션은 자동 추출을 지원합니다.

데이터 타입으로 암시적으로 매핑할 수 있는 클래스의 전체 목록은 데이터 타입 추출 섹션을 참조하세요.

@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;

// function with overloaded evaluation methods
public static class OverloadedFunction extends ScalarFunction {

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

  // define the precision and scale of a decimal
  public @DataTypeHint("DECIMAL(12, 3)") BigDecimal eval(double a, double b) {
    return BigDecimal.valueOf(a + b);
  }

  // define a nested data type
  @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
  public Row eval(int i) {
    return Row.of(String.valueOf(i), Instant.ofEpochSecond(i));
  }

  // allow wildcard input and customly serialized output
  @DataTypeHint(value = "RAW", bridgedTo = ByteBuffer.class)
  public ByteBuffer eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
    return MyUtils.serializeToByteBuffer(o);
  }
}

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
import scala.annotation.varargs

// function with overloaded evaluation methods
class OverloadedFunction extends ScalarFunction {

  // no hint required
  def eval(a: Long, b: Long): Long = {
    a + b
  }

  // define the precision and scale of a decimal
  @DataTypeHint("DECIMAL(12, 3)")
  def eval(a: Double, b: Double): BigDecimal = {
    BigDecimal(a + b)
  }

  // define a nested data type
  @DataTypeHint("ROW<s STRING, t TIMESTAMP_LTZ(3)>")
  def eval(i: Int): Row = {
    Row.of(i.toString, java.time.Instant.ofEpochSecond(i))
  }

  // allow wildcard input and customly serialized output
  @DataTypeHint(value = "RAW", bridgedTo = classOf[java.nio.ByteBuffer])
  def eval(@DataTypeHint(inputGroup = InputGroup.ANY) o: Any): java.nio.ByteBuffer = {
    MyUtils.serializeToByteBuffer(o)
  }
}

@FunctionHint

일부 시나리오에서는 하나의 평가 메서드가 동시에 여러 다른 데이터 타입을 처리하는 것이 바람직합니다. 또한 일부 시나리오에서는 오버로드된 평가 메서드가 공통 결과 타입을 갖는데, 이는 한 번만 선언되어야 합니다.

@FunctionHint 어노테이션은 인자 데이터 타입에서 결과 데이터 타입으로의 매핑을 제공할 수 있습니다. 전체 함수 클래스나 평가 메서드에 입력, 축적자(accumulator), 결과 데이터 타입을 어노테이션할 수 있습니다. 클래스 위에 여러 어노테이션을 선언하거나 함수 시그니처를 오버로드하기 위해 각 평가 메서드별로 하나씩 선언할 수 있습니다. 모든 힌트 파라미터는 선택 사항입니다. 파라미터가 정의되지 않으면 기본 리플렉션 기반 추출이 사용됩니다. 함수 클래스 위에 정의된 힌트 파라미터는 모든 평가 메서드에 상속됩니다.

다음 예제는 함수 힌트를 사용하는 방법을 보여줍니다. 더 자세한 정보는 어노테이션 클래스 문서에서 확인할 수 있습니다.

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;

// function with overloaded evaluation methods
// but globally defined output type
@FunctionHint(output = @DataTypeHint("ROW<s STRING, i INT>"))
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));
  }
}

// decouples the type inference from evaluation methods,
// the type inference is entirely determined by the function hints
@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]);
  }
}

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

// function with overloaded evaluation methods
// but globally defined output type
@FunctionHint(output = new DataTypeHint("ROW<s STRING, i INT>"))
class OverloadedFunction extends TableFunction[Row] {

  def eval(a: Int, b: Int): Unit = {
    collect(Row.of("Sum", Int.box(a + b)))
  }

  // overloading of arguments is still possible
  def eval(): Unit = {
    collect(Row.of("Empty args", Int.box(-1)))
  }
}

// decouples the type inference from evaluation methods,
// the type inference is entirely determined by the function hints
@FunctionHint(
  input = Array(new DataTypeHint("INT"), new DataTypeHint("INT")),
  output = new DataTypeHint("INT")
)
@FunctionHint(
  input = Array(new DataTypeHint("BIGINT"), new DataTypeHint("BIGINT")),
  output = new DataTypeHint("BIGINT")
)
@FunctionHint(
  input = Array(),
  output = new DataTypeHint("BOOLEAN")
)
class OverloadedFunction extends TableFunction[AnyRef] {

  // an implementer just needs to make sure that a method exists
  // that can be called by the JVM
  @varargs
  def eval(o: AnyRef*) = {
    if (o.length == 0) {
      collect(Boolean.box(false))
    }
    collect(o(0))
  }
}
사용자 정의 타입 추론 (Custom Type Inference)

대부분의 시나리오에서 @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;
    }
  }

  // the automatic, reflection-based type inference is disabled and
  // replaced by the following logic
  @Override
  public TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()
      // specify typed arguments
      // parameters will be casted implicitly to those types if necessary
      .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.");
        }
        // return a data type based on a literal
        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();
  }
}

사용자 정의 타입 추론의 더 많은 예는 flink-examples-table 모듈의 고급 함수 구현을 참조하세요.

명명된 파라미터 (Named Parameters)

함수를 호출할 때 파라미터 이름을 사용해 파라미터의 값을 지정할 수 있습니다. 명명된 파라미터(named parameters)는 파라미터 이름과 값을 함께 함수에 전달할 수 있게 해주며, 잘못된 파라미터 순서로 인한 혼란을 피하고 코드의 가독성과 유지보수성을 높입니다. 또한 명명된 파라미터는 선택적 파라미터를 생략할 수 있는데, 생략하면 기본적으로 null로 채워집니다.

@ArgumentHint 어노테이션을 사용해 파라미터의 이름, 타입, 필수 여부를 지정할 수 있습니다.

다음 3가지 예제는 서로 다른 범위에서 @ArgumentHint를 사용하는 방법을 보여줍니다. 더 자세한 정보는 어노테이션 클래스 문서에서 확인할 수 있습니다.

  1. 함수의 eval 메서드 파라미터에 @ArgumentHint 어노테이션 사용하기
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;

public static class NamedParameterClass extends ScalarFunction {

    // Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
    public String eval(@ArgumentHint(name = "param1", isOptional = false, type = @DataTypeHint("STRING")) String s1,
                       @ArgumentHint(name = "param2", isOptional = true, type = @DataTypeHint("INT")) Integer s2) {
        return s1 + ", " + s2;
    }
}

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;

class NamedParameterClass extends ScalarFunction {

  // Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
  def eval(@ArgumentHint(name = "param1", isOptional = false, `type` = new DataTypeHint("STRING")) s1: String,
           @ArgumentHint(name = "param2", isOptional = true, `type` = new DataTypeHint("INTEGER")) s2: Integer) = {
    s1 + ", " + s2
  }
}
  1. 함수의 eval 메서드에 @ArgumentHint 어노테이션 사용하기
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.functions.ScalarFunction;

public static class NamedParameterClass extends ScalarFunction {

  // Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
  @FunctionHint(
    arguments = {
      @ArgumentHint(name = "param1", isOptional = false, type = @DataTypeHint("STRING")),
      @ArgumentHint(name = "param2", isOptional = true, type = @DataTypeHint("INTEGER"))
    }
  )
  public String eval(String s1, Integer s2) {
    return s1 + ", " + s2;
  }
}

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.ScalarFunction;

class NamedParameterClass extends ScalarFunction {

  // Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
  @FunctionHint(
    arguments = Array(
      new ArgumentHint(name = "param1", isOptional = false, `type` = new DataTypeHint("STRING")),
      new ArgumentHint(name = "param2", isOptional = true, `type` = new DataTypeHint("INTEGER"))
    )
  )
  def eval(s1: String, s2: Int): String = {
    s1 + ", " + s2
  }
}
  1. 함수의 클래스에 @ArgumentHint 어노테이션 사용하기
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.ScalarFunction;

// Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
@FunctionHint(
  arguments = {
    @ArgumentHint(name = "param1", isOptional = false, type = @DataTypeHint("STRING")),
    @ArgumentHint(name = "param2", isOptional = true, type = @DataTypeHint("INTEGER"))
  }
)
public static class NamedParameterClass extends ScalarFunction {

  public String eval(String s1, Integer s2) {
    return s1 + ", " + s2;
  }
}

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.ScalarFunction;

// Use the @ArgumentHint annotation to specify the name, type, and whether a parameter is required.
@FunctionHint(
  arguments = Array(
    new ArgumentHint(name = "param1", isOptional = false, `type` = new DataTypeHint("STRING")),
    new ArgumentHint(name = "param2", isOptional = true, `type` = new DataTypeHint("INTEGER"))
  )
)
class NamedParameterClass extends ScalarFunction {

  def eval(s1: String, s2: Int): String = {
    s1 + ", " + s2
  }
}
  • @ArgumentHint 어노테이션은 이미 @DataTypeHint 어노테이션을 포함하므로 @FunctionHint 안에서 @DataTypeHint와 함께 사용할 수 없습니다. 함수 파라미터에 적용할 때 @ArgumentHint는 @DataTypeHint와 동시에 사용할 수 없으며, @ArgumentHint 사용을 권장합니다.
  • 명명된 파라미터는 해당 클래스에 오버로드된 함수와 가변 인자 함수가 없을 때만 효과가 있습니다. 그렇지 않으면 명명된 파라미터를 사용하면 오류가 발생합니다.

결정성 (Determinism)

모든 사용자 정의 함수 클래스는 isDeterministic() 메서드를 오버라이드해 결정적(deterministic) 결과를 생성하는지 여부를 선언할 수 있습니다. 함수가 순수 함수형이 아니라면(random(), date(), now()처럼) 메서드는 false를 반환해야 합니다. 기본적으로 isDeterministic()은 true를 반환합니다.

또한 isDeterministic() 메서드는 런타임 동작에도 영향을 줄 수 있습니다. 런타임 구현은 두 가지 다른 단계에서 호출될 수 있습니다.

  1. 플래닝 중(즉 pre-flight 단계): 함수가 상수 표현식과 함께 호출되거나 주어진 문장에서 상수 표현식을 도출할 수 있으면, 상수 표현식 축소를 위해 함수가 사전 평가되며 더 이상 클러스터에서 실행되지 않을 수 있습니다. 이 경우 isDeterministic()이 상수 표현식 축소를 비활성화하는 데 사용되지 않는 한 그렇습니다. 예를 들어 다음 ABS 호출은 플래닝 중에 실행됩니다: SELECT ABS(-1) FROM t와 SELECT ABS(field) FROM t WHERE field = -1. 반면 SELECT ABS(field) FROM t는 그렇지 않습니다.
  2. 런타임 중(즉 클러스터 실행): 함수가 비상수 표현식과 함께 호출되거나 isDeterministic()이 false를 반환하는 경우.
시스템(내장) 함수 결정성 (System (Built-in) Function Determinism)

시스템(내장) 함수의 결정성은 변경할 수 없습니다. Apache Calcite의 SqlOperator 정의에 따라 결정적이지 않은 두 종류의 함수가 있습니다: 동적 함수(dynamic function)와 비결정적 함수(non-deterministic function).

  /**
   * 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;
  }

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은 시스템 함수에만 적용됩니다.

상수 표현식 축소 (Constant Expression Reduction)

사용자 정의 함수는 supportsConstantFolding() 메서드를 오버라이드해 상수 표현식 축소를 허용하는지 여부를 선언할 수 있습니다. 상수 인자를 가진 함수 호출은 경우에 따라 축소되고 단순화될 수 있습니다. 예를 들어 사용자 정의 함수 호출 PlusOne(10)은 표현식에서 단순히 11로 단순화될 수 있습니다. 이 최적화는 플래닝 시간에 발생하며 축소된 값만 사용하는 플랜을 만듭니다. 이것은 일반적으로 바람직하므로 기본적으로 활성화되어 있지만, 비활성화해야 하는 경우도 있습니다.

한 가지 경우는 함수 호출이 결정적이지 않은 경우입니다. 이는 위의 결정성 섹션에서 더 자세히 다룹니다. 함수를 비결정적으로 설정하면 supportsConstantFolding()이 true여도 함수 호출 표현식 축소를 방지하는 효과가 있습니다.

함수 호출은 항상 결정적 결과를 반환하더라도 부작용(side effect)이 있을 수 있습니다. 이는 Flink 내 쿼리의 정확성은 상수 표현식 축소를 허용할 수 있지만, 그래도 바람직하지 않을 수 있음을 의미합니다. 이 경우 supportsConstantFolding() 메서드를 false를 반환하도록 설정하면 상수 표현식 축소를 방지하고 런타임 호출을 보장하는 효과도 있습니다.

런타임 통합 (Runtime Integration)

때로는 사용자 정의 함수가 실제 작업 전에 전역 런타임 정보를 얻거나 일부 설정/정리 작업을 수행해야 할 수 있습니다. 사용자 정의 함수는 오버라이드할 수 있는 open()과 close() 메서드를 제공하며, DataStream API의 RichFunction 메서드와 유사한 기능을 제공합니다.

open() 메서드는 평가 메서드 전에 한 번 호출됩니다. close() 메서드는 평가 메서드의 마지막 호출 이후에 호출됩니다.

open() 메서드는 사용자 정의 함수가 실행되는 컨텍스트에 대한 정보(예: 메트릭 그룹, 분산 캐시 파일, 전역 작업 파라미터)를 포함하는 FunctionContext를 제공합니다.

FunctionContext의 해당 메서드를 호출해 다음 정보를 얻을 수 있습니다.

메서드 (Method) 설명 (Description)
getMetricGroup() 이 병렬 서브태스크의 메트릭 그룹.
getCachedFile(name) 분산 캐시 파일의 로컬 임시 파일 복사본.
getJobParameter(name, defaultValue) 주어진 키와 연관된 전역 작업 파라미터 값.
getExternalResourceInfos(resourceName) 주어진 키와 연관된 외부 리소스 정보의 집합을 반환.

참고: 함수가 실행되는 컨텍스트에 따라 위의 모든 메서드를 사용할 수 없을 수 있습니다. 예를 들어 상수 표현식 축소 중에는 메트릭을 추가하는 것이 no-op 작업입니다.

다음 예제 스니펫은 전역 작업 파라미터에 접근하기 위해 스칼라 함수에서 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 {
        // access the global "hashcode_factor" parameter
        // "12" would be the default value if the parameter does not exist
        factor = Integer.parseInt(context.getJobParameter("hashcode_factor", "12"));
    }

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

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

// add job parameter
env.getConfig().addJobParameter("hashcode_factor", "31");

// register the function
env.createTemporarySystemFunction("hashCode", HashCodeFunction.class);

// use the function
env.sqlQuery("SELECT myField, hashCode(myField) FROM MyTable");

import org.apache.flink.table.api._
import org.apache.flink.table.functions.FunctionContext
import org.apache.flink.table.functions.ScalarFunction

class HashCodeFunction extends ScalarFunction {

  private var factor: Int = 0

  override def open(context: FunctionContext): Unit = {
    // access the global "hashcode_factor" parameter
    // "12" would be the default value if the parameter does not exist
    factor = context.getJobParameter("hashcode_factor", "12").toInt
  }

  def eval(s: String): Int = {
    s.hashCode * factor
  }
}

val env = TableEnvironment.create(...)

// add job parameter
env.getConfig.addJobParameter("hashcode_factor", "31")

// register the function
env.createTemporarySystemFunction("hashCode", classOf[HashCodeFunction])

// use the function
env.sqlQuery("SELECT myField, hashCode(myField) FROM MyTable")

스칼라 함수 (Scalar Functions)

사용자 정의 스칼라 함수는 0개, 1개 또는 여러 개의 스칼라 값을 새로운 스칼라 값으로 매핑합니다. 데이터 타입 섹션에 나열된 모든 데이터 타입을 평가 메서드의 파라미터 또는 반환 타입으로 사용할 수 있습니다.

스칼라 함수를 정의하려면 org.apache.flink.table.functions의 ScalarFunction 기본 클래스를 확장하고 eval(...)이라는 이름의 평가 메서드 하나 이상을 구현해야 합니다.

다음 예제는 자체 해시 코드 함수를 정의하고 쿼리에서 호출하는 방법을 보여줍니다. 자세한 내용은 구현 가이드를 참조하세요.

import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.ScalarFunction;
import static org.apache.flink.table.api.Expressions.*;

public static class HashFunction extends ScalarFunction {

  // take any data type and return INT
  public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
    return o.hashCode();
  }
}

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

// call function "inline" without registration in Table API
env.from("MyTable").select(call(HashFunction.class, $("myField")));

// register function
env.createTemporarySystemFunction("HashFunction", HashFunction.class);

// call registered function in Table API
env.from("MyTable").select(call("HashFunction", $("myField")));

// call registered function in SQL
env.sqlQuery("SELECT HashFunction(myField) FROM MyTable");

import org.apache.flink.table.annotation.InputGroup
import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction

class HashFunction extends ScalarFunction {

  // take any data type and return INT
  def eval(@DataTypeHint(inputGroup = InputGroup.ANY) o: AnyRef): Int = {
    o.hashCode()
  }
}

val env = TableEnvironment.create(...)

// call function "inline" without registration in Table API
env.from("MyTable").select(call(classOf[HashFunction], $"myField"))

// register function
env.createTemporarySystemFunction("HashFunction", classOf[HashFunction])

// call registered function in Table API
env.from("MyTable").select(call("HashFunction", $"myField"))

// call registered function in SQL
env.sqlQuery("SELECT HashFunction(myField) FROM MyTable")

Python에서 함수를 구현하거나 호출하려면 Python 스칼라 함수 문서를 참조하세요.

비동기 스칼라 함수 (Asynchronous Scalar Functions)

외부 시스템과 상호작용할 때(예: 데이터베이스에 저장된 데이터로 스트림 이벤트를 보강할 때), 네트워크나 기타 지연이 스트리밍 애플리케이션의 실행 시간을 지배하지 않도록 주의해야 합니다.

예를 들어 ScalarFunction을 사용해 외부 데이터베이스의 데이터에 순진하게 접근하는 것은 일반적으로 동기적 상호작용을 의미합니다: 데이터베이스에 요청을 보내고 ScalarFunction은 응답을 받을 때까지 기다립니다. 많은 경우 이 대기가 함수 시간의 대부분을 차지합니다.

이 비효율성을 해결하기 위해 AsyncScalarFunction이 있습니다. 데이터베이스와의 비동기 상호작용은 단일 함수 인스턴스가 많은 요청을 동시에 처리하고 응답을 동시에 받을 수 있음을 의미합니다. 이렇게 하면 대기 시간을 다른 요청 전송 및 응답 수신과 겹칠 수 있습니다. 최소한 대기 시간은 여러 요청에 걸쳐 상각됩니다. 이는 대부분의 경우 훨씬 더 높은 스트리밍 처리량으로 이어집니다.

AsyncScalarFunction 정의 (Defining an AsyncScalarFunction)

사용자 정의 비동기 스칼라 함수는 0개, 1개 또는 여러 개의 스칼라 값을 새로운 스칼라 값으로 매핑합니다. 데이터 타입 섹션에 나열된 모든 데이터 타입을 평가 메서드의 파라미터 또는 반환 타입으로 사용할 수 있습니다.

비동기 스칼라 함수를 정의하려면 org.apache.flink.table.functions의 AsyncScalarFunction 기본 클래스를 확장하고 eval(...)이라는 이름의 평가 메서드 하나 이상을 구현해야 합니다. 첫 번째 인자는 결과를 반환하는 데 사용되는 CompletableFuture<...>여야 하며, 그 뒤의 인자는 함수에 전달된 파라미터입니다.

eval에 대한 미해결(outstanding) 호출 수는 table.exec.async-scalar.max-concurrent-operations로 구성할 수 있습니다.

다음 예제는 백그라운드의 스레드 풀에서 작업을 수행하는 방법을 보여줍니다. 다만 비동기 인터페이스를 노출하는 모든 라이브러리를 콜백에서 CompletableFuture를 완성하는 데 직접 사용할 수 있습니다. 자세한 내용은 구현 가이드를 참조하세요.

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.AsyncScalarFunction;

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import static org.apache.flink.table.api.Expressions.*;

/**
 * A function which simulates looking up a beverage name from a database.
 * Since such lookups are often slow, we use an AsyncScalarFunction.
 */
public static class BeverageNameLookupFunction extends AsyncScalarFunction {
    private transient Executor executor;

    @Override
    public void open(FunctionContext context) {
        // Create a thread pool for executing the background lookup.
        executor = Executors.newFixedThreadPool(10);
    }

    // The eval method takes a future for the result and the beverage ID to lookup.
    public void eval(CompletableFuture<String> future, Integer beverageId) {
        // Submit a task to the thread pool. We don't want to block this main
        // thread since that would prevent concurrent execution. The future can be
        // completed from another thread when the lookup is done.
        executor.execute(() -> {
            // Simulate a database lookup by sleeping for 1s.
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {}
            // Complete the future with the right beverage name.
            switch (beverageId) {
                case 0:
                    future.complete("Latte");
                    break;
                case 1:
                    future.complete("Cappuccino");
                    break;
                case 2:
                    future.complete("Espresso");
                    break;
                default:
                    // In the exceptional case, return an error.
                    future.completeExceptionally(new IllegalArgumentException("Bad beverageId: " + beverageId));
            }
        });
    }
}

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

env.getConfig().set("table.exec.async-scalar.max-concurrent-operations", "5");
env.getConfig().set("table.exec.async-scalar.timeout", "1m");

// call function "inline" without registration in Table API
env.from("Beverages").select(call(BeverageNameLookupFunction.class, $("beverageId")));

// register function
env.createTemporarySystemFunction("GetBeverageName", BeverageNameLookupFunction.class);

// call registered function in Table API
env.from("Beverages").select(call("GetBeverageName", $("beverageId")));

// call registered function in SQL
env.sqlQuery("SELECT GetBeverageName(beverageId) FROM Beverages");

비동기 의미 (Asynchronous Semantics)

AsyncScalarFunction에 대한 호출은 원래 입력 순서와 다르게 완료될 수 있지만, 올바른 의미를 유지하기 위해 함수의 출력은 쿼리의 다운스트림 컴포넌트에 그 입력 순서를 유지하는 것이 보장됩니다. 데이터 자체가 완료 순서를 드러낼 수 있으므로(예: 가져온 타임스탬프 포함), 사용자는 이것이 자신의 use-case에 허용 가능한지 고려해야 합니다.

오류 처리 (Error Handling)

사용자가 오류를 나타내는 기본 방법은 CompletableFuture.completeExceptionally(Throwable)을 호출하는 것입니다. 마찬가지로 eval 호출 시 시스템이 예외를 만나면 그것도 오류가 됩니다. 오류가 발생하면 시스템은 table.exec.async-scalar.retry-strategy로 구성된 재시도 전략을 고려합니다. 이것이 NO_RETRY이면 작업이 실패합니다. FIXED_DELAY로 설정되면 table.exec.async-scalar.retry-delay만큼 기다린 후 함수 호출을 재시도합니다. table.exec.async-scalar.max-attempts만큼 실패했거나 타임아웃 table.exec.async-scalar.timeout(모든 재시도 시도 포함)이 만료되면 작업이 실패합니다.

AsyncScalarFunction vs. ScalarFunction

UDF가 블로킹 호출이 없는 CPU 집약적 로직을 포함하는지 고려해야 합니다. 그렇다면 비동기 기능이 필요하지 않을 것이므로 ScalarFunction을 사용할 수 있습니다. 로직이 네트워크나 백그라운드 작업(예: 데이터베이스 룩업, RPC, REST 호출)을 기다리는 것을 포함한다면 이것이 속도를 높이는 유용한 방법이 될 수 있습니다. 또한 AsyncScalarFunction을 지원하지 않는 쿼리도 있으므로, 확실하지 않으면 ScalarFunction을 사용해야 합니다.

테이블 함수 (Table Functions)

사용자 정의 스칼라 함수와 유사하게, 사용자 정의 테이블 함수(UDTF)는 0개, 1개 또는 여러 개의 스칼라 값을 입력 인자로 받습니다. 그러나 단일 값을 출력하는 대신 임의의 수의 행(또는 구조화된 타입)을 출력으로 반환할 수 있습니다. 반환된 레코드는 하나 이상의 필드로 구성될 수 있습니다. 출력 레코드가 단일 필드로만 구성되면 구조화된 레코드를 생략하고 런타임이 암시적으로 행으로 감싸는 스칼라 값을 방출할 수 있습니다.

테이블 함수를 정의하려면 org.apache.flink.table.functions의 TableFunction 기본 클래스를 확장하고 eval(...)이라는 이름의 평가 메서드 하나 이상을 구현해야 합니다. 다른 함수와 유사하게 입력 및 출력 데이터 타입은 리플렉션을 사용해 자동으로 추출됩니다. 여기에는 출력 데이터 타입을 결정하기 위한 클래스의 제네릭 인자 T가 포함됩니다. 스칼라 함수와 달리 평가 메서드 자체는 반환 타입을 가지지 않아야 하며, 대신 테이블 함수는 각 평가 메서드 안에서 0개, 1개 또는 여러 개의 레코드를 방출하기 위해 호출할 수 있는 collect(T) 메서드를 제공합니다.

Table API에서 테이블 함수는 .joinLateral(...) 또는 .leftOuterJoinLateral(...)과 함께 사용됩니다. joinLateral 연산자는 외부 테이블(연산자 왼쪽의 테이블)의 각 행을 테이블 반환 함수(연산자 오른쪽)가 생성한 모든 행과 (크로스) 조인합니다. leftOuterJoinLateral 연산자는 외부 테이블의 각 행을 테이블 반환 함수가 생성한 모든 행과 조인하면서, 테이블 함수가 빈 테이블을 반환하는 외부 행을 보존합니다.

SQL에서는 JOIN 또는 LEFT JOIN과 ON TRUE 조인 조건으로 LATERAL TABLE(<TableFunction>)을 사용합니다.

다음 예제는 자체 split 함수를 정의하고 쿼리에서 호출하는 방법을 보여줍니다. 자세한 내용은 구현 가이드를 참조하세요.

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.*;

@FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
public static class SplitFunction extends TableFunction<Row> {

  public void eval(String str) {
    for (String s : str.split(" ")) {
      // use collect(...) to emit a row
      collect(Row.of(s, s.length()));
    }
  }
}

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

// call function "inline" without registration in Table API
env
  .from("MyTable")
  .joinLateral(call(SplitFunction.class, $("myField")))
  .select($("myField"), $("word"), $("length"));

env
  .from("MyTable")
  .leftOuterJoinLateral(call(SplitFunction.class, $("myField")))
  .select($("myField"), $("word"), $("length"));

// rename fields of the function in Table API
env
  .from("MyTable")
  .leftOuterJoinLateral(call(SplitFunction.class, $("myField")).as("newWord", "newLength"))
  .select($("myField"), $("newWord"), $("newLength"));

// register function
env.createTemporarySystemFunction("SplitFunction", SplitFunction.class);

// call registered function in Table API
env
  .from("MyTable")
  .joinLateral(call("SplitFunction", $("myField")))
  .select($("myField"), $("word"), $("length"));

env
  .from("MyTable")
  .leftOuterJoinLateral(call("SplitFunction", $("myField")))
  .select($("myField"), $("word"), $("length"));

// call registered function in SQL
env.sqlQuery("SELECT myField, word, length FROM MyTable, LATERAL TABLE(SplitFunction(myField))");
env.sqlQuery("SELECT myField, word, length FROM MyTable LEFT JOIN LATERAL TABLE(SplitFunction(myField)) ON TRUE");

Python에서 함수를 구현하거나 호출하려면 Python 테이블 함수 문서를 참조하세요.

비동기 테이블 함수 (Asynchronous Table Functions)

AsyncScalarFunction과 유사하게, 단일 스칼라 값 대신 여러 행 결과를 반환하기 위한 AsyncTableFunction도 존재합니다. 마찬가지로 이것은 외부 시스템과 상호작용할 때(예: 데이터베이스에 저장된 데이터로 스트림 이벤트를 보강할 때) 가장 유용합니다.

외부 시스템과의 비동기 상호작용은 단일 함수 인스턴스가 많은 요청을 동시에 처리하고 응답을 동시에 받을 수 있음을 의미합니다. 이렇게 하면 대기 시간을 다른 요청 전송 및 응답 수신과 겹칠 수 있습니다. 최소한 대기 시간은 여러 요청에 걸쳐 상각됩니다. 이는 대부분의 경우 훨씬 더 높은 스트리밍 처리량으로 이어집니다.

AsyncTableFunction 정의 (Defining an AsyncTableFunction)

비동기 테이블 함수를 정의하려면 org.apache.flink.table.functions의 AsyncTableFunction 기본 클래스를 확장하고 eval(...)이라는 이름의 평가 메서드 하나 이상을 구현해야 합니다. 첫 번째 인자는 결과를 반환하는 데 사용되는 CompletableFuture<...>여야 하며, 그 뒤의 인자는 함수에 전달된 파라미터입니다.

비동기 의미 (Asynchronous Semantics)

AsyncTableFunction에 대한 호출은 원래 입력 순서와 다르게 완료될 수 있지만, 올바른 의미를 유지하기 위해 함수의 출력은 쿼리의 다운스트림 컴포넌트에 그 입력 순서를 유지하는 것이 보장됩니다. 데이터 자체가 완료 순서를 드러낼 수 있으므로 사용자는 이것이 자신의 use-case에 허용 가능한지 고려해야 합니다.

다음 예제는 HTTP 요청을 비동기적으로 처리하는 방법을 보여줍니다. 자세한 내용은 구현 가이드를 참조하세요.

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.AsyncTableFunction;
import org.apache.flink.types.Row;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;
import static org.apache.flink.table.api.Expressions.*;

@FunctionHint(
  output = @DataTypeHint("ROW<user STRING, tour STRING>")
)
public static class AsyncBackgroundFunction extends AsyncTableFunction<Row> {

  @Override
  public void eval(CompletableFuture<Row> future, String userId) {
    future.complete(Row.of(userId, "tour"));
  }
}

Table API 사용 예 (스칼라 함수와 동일한 패턴):

// call function "inline" without registration in Table API
env.from("MyTable").joinLateral(call(AsyncBackgroundFunction.class, $("userId")));

// register function
env.createTemporarySystemFunction("AsyncBackgroundFunction", AsyncBackgroundFunction.class);

// call registered function in Table API
env.from("MyTable").joinLateral(call("AsyncBackgroundFunction", $("userId")));

// call registered function in SQL
env.sqlQuery("SELECT * FROM MyTable LEFT JOIN LATERAL TABLE(AsyncBackgroundFunction(userId)) ON TRUE");

Python에서 함수를 구현하거나 호출하려면 Python 테이블 함수 문서를 참조하세요.

집계 함수 (Aggregate Functions)

사용자 정의 집계 함수(UDAGG)는 여러 행의 스칼라 값을 새로운 스칼라 값으로 매핑합니다.

집계 함수의 동작은 축적자(accumulator)의 개념을 중심으로 합니다. 축적자는 최종 집계 결과가 계산될 때까지 집계된 값을 저장하는 중간 데이터 구조입니다.

각 집계할 행 집합에 대해 런타임은 createAccumulator()를 호출해 빈 축적자를 만듭니다. 그 후 함수의 accumulate(...) 메서드가 각 입력 행에 대해 호출되어 축적자를 업데이트합니다. 모든 행이 처리되면 함수의 getResult(...) 메서드가 호출되어 최종 결과를 계산하고 반환합니다.

다음 예제는 집계 과정을 보여줍니다. 이 예제는 id, name, price 세 컬럼과 5개의 행을 가진 음료 데이터 테이블을 가정합니다. 테이블에서 모든 음료 중 가장 높은 가격 2개를 찾고 싶다고 가정해 보겠습니다(즉 Top2() 테이블 집계 수행). 5개의 행 각각을 고려해야 합니다. 예제에서 WeightedAvg 함수는 weightedAvg라는 이름으로 TableEnvironment에 등록되며, 집계 함수는 주어진 입력 값의 가중 평균을 계산합니다.

집계 함수는 accumulate() 메서드의 인자로 ACC(축적자) 타입과 사용자 정의 입력 값을 정의합니다. createAccumulator() 메서드는 빈 축적자를 만들고, accumulate(...) 메서드는 각 입력 행으로 축적자를 업데이트하며, getResult(...) 메서드는 축적자에서 최종 결과를 반환합니다. retract(...) 메서드는 축적자에서 입력 값의 기여를 취소(철회)하는 선택적 메서드입니다.

import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.AggregateFunction;
import static org.apache.flink.table.api.Expressions.*;

// mutable accumulator of structured type for the aggregate function
public static class WeightedAvgAccumulator {
  public long sum = 0;
  public int count = 0;
}

// function that takes (Long INT, INT) and returns Long
public static class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccumulator> {

  @Override
  public WeightedAvgAccumulator createAccumulator() {
    return new WeightedAvgAccumulator();
  }

  public void accumulate(WeightedAvgAccumulator acc, Long iValue, int iWeight) {
    acc.sum += iValue * iWeight;
    acc.count += iWeight;
  }

  public void retract(WeightedAvgAccumulator acc, Long iValue, int iWeight) {
    acc.sum -= iValue * iWeight;
    acc.count -= iWeight;
  }

  @Override
  public Long getValue(WeightedAvgAccumulator acc) {
    if (acc.count == 0) {
      return null;
    } else {
      return acc.sum / acc.count;
    }
  }
}

TableEnvironment env = TableEnvironment.create(...);
env.executeSql("CREATE TABLE MyTable (myField BIGINT, weight INT) WITH (...)"); // define table
env.createTemporarySystemFunction("WeightedAvg", WeightedAvg.class);

select call registered function in Table API / SQL:
env.from("MyTable")
   .groupBy($("myField"))
   .select($("myField"), call("WeightedAvg", $("myField"), $("weight")));
env.sqlQuery("SELECT MyField, WeightedAvg(myField, weight) FROM MyTable GROUP BY MyField");

집계 함수의 구현 가이드와 집계 함수 문서에서 자세한 내용을 확인할 수 있습니다.

필수 및 선택 메서드 (Mandatory and Optional Methods)

각 AggregateFunction에 대해 다음 메서드는 필수입니다.

  • createAccumulator()
  • accumulate(...)
  • getValue(...)

추가로 선택적으로 구현할 수 있는 몇 가지 메서드가 있습니다. 일부 메서드는 시스템이 더 효율적인 쿼리 실행을 허용하고, 다른 일부는 특정 use-case에 필수입니다. 예를 들어 merge(...) 메서드는 집계 함수가 세션 그룹 윈도우 컨텍스트에서 적용되어야 하는 경우 필수입니다(두 세션 윈도우의 축적자를 연결하는 행이 관찰될 때 결합해야 하기 때문).

AggregateFunction의 다음 메서드는 use-case에 따라 필요합니다.

  • retract(...)는 OVER 윈도우에서의 집계에 필요합니다.
  • merge(...)는 많은 유한 집계와 무한 세션 및 홉 윈도우 집계에 필요합니다.

accumulate(...) 메서드의 상세:

/*
 * Processes the input values and updates the provided accumulator instance. The method
 * accumulate can be overloaded with different custom types and arguments. An aggregate function
 * requires at least one accumulate() method.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: [user defined inputs] the input value (usually obtained from new arrived data).
 */
public void accumulate(ACC accumulator, [user defined inputs])

retract(...) 메서드의 상세:

/*
 * Retracts the input values from the accumulator instance. The current design assumes the
 * inputs are the values that have been previously accumulated. The method retract can be
 * overloaded with different custom types and arguments. This method must be implemented for
 * bounded OVER aggregates over unbounded tables.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: [user defined inputs] the input value (usually obtained from new arrived data).
 */
public void retract(ACC accumulator, [user defined inputs])

merge(...) 메서드의 상세:

/*
 * Merges a group of accumulator instances into one accumulator instance. This method must be
 * implemented for unbounded session window grouping aggregates and bounded grouping aggregates.
 *
 * param: accumulator the accumulator which will keep the merged aggregate results. It should
 *                    be noted that the accumulator may contain the previous aggregated
 *                    results. Therefore user should not replace or clean this instance in the
 *                    custom merge method.
 * param: iterable    an java.lang.Iterable pointed to a group of accumulators that will be
 *                    merged.
 */
public void merge(ACC accumulator, java.lang.Iterable<ACC> iterable)

Python에서 함수를 구현하거나 호출하려면 Python 집계 함수 문서를 참조하세요.

테이블 집계 함수 (Table Aggregate Functions)

사용자 정의 테이블 집계 함수(UDTAGG)는 여러 행의 스칼라 값을 0개, 1개 또는 여러 개의 행(또는 구조화된 타입)으로 매핑합니다. 반환된 레코드는 하나 이상의 필드로 구성될 수 있습니다. 출력 레코드가 단일 필드로만 구성되면 구조화된 레코드를 생략하고, 런타임이 암시적으로 행으로 감싸는 스칼라 값을 방출할 수 있습니다.

집계 함수와 유사하게 테이블 집계의 동작은 축적자(accumulator)의 개념을 중심으로 합니다. 축적자는 최종 집계 결과가 계산될 때까지 집계된 값을 저장하는 중간 데이터 구조입니다.

집계해야 하는 각 행 집합에 대해 런타임은 createAccumulator()를 호출해 빈 축적자를 만듭니다. 그 후 함수의 accumulate(...) 메서드가 각 입력 행에 대해 호출되어 축적자를 업데이트합니다. 모든 행이 처리되면 함수의 emitValue(...) 또는 emitUpdateWithRetract(...) 메서드가 호출되어 최종 결과를 계산하고 반환합니다.

다음 예제는 집계 과정을 보여줍니다. 예제에서 우리는 좌표를 나타내는 태양 사건들을 처리하는 함수를 정의합니다. 예제는 id, name, price 세 컬럼과 5개의 행을 가진 음료 데이터 테이블을 가정합니다. 테이블에서 모든 음료 중 가장 높은 가격 2개(즉 TOP2() 테이블 집계 수행)를 찾고 싶다고 가정해 보겠습니다. 5개의 행 각각을 고려해야 합니다. 결과는 상위 2개 값을 가진 테이블입니다.

테이블 집계 함수를 정의하려면 org.apache.flink.table.functions의 TableAggregateFunction 기본 클래스를 확장하고 accumulate(...)라는 이름의 평가 메서드 하나 이상을 구현해야 합니다. accumulate 메서드는 public으로 선언되어야 하며 static이 아니어야 합니다. accumulate 메서드는 accumulate라는 이름의 여러 메서드를 구현해 오버로드할 수도 있습니다.

기본적으로 입력, 축적자, 출력 데이터 타입은 리플렉션을 사용해 자동으로 추출됩니다. 여기에는 축적자 데이터 타입을 결정하기 위한 클래스의 제네릭 인자 ACC와 축적자 데이터 타입을 결정하기 위한 제네릭 인자 T가 포함됩니다. 입력 인자는 하나 이상의 accumulate(...) 메서드에서 도출됩니다. 자세한 내용은 구현 가이드를 참조하세요.

Python에서 함수를 구현하거나 호출하려면 Python 함수 문서를 참조하세요.

다음 예제는 자체 테이블 집계 함수를 정의하고 쿼리에서 호출하는 방법을 보여줍니다.

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.*;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;
import static org.apache.flink.table.api.Expressions.*;

// mutable accumulator of structured type for the aggregate function
public static class Top2Accumulator {
  public Integer first;
  public Integer second;
}

// function that takes (value INT), stores intermediate results in a structured
// type of Top2Accumulator, and returns the result as a structured type of Tuple2<Integer, Integer>
// for value and rank
public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accumulator> {

  @Override
  public Top2Accumulator createAccumulator() {
    Top2Accumulator acc = new Top2Accumulator();
    acc.first = Integer.MIN_VALUE;
    acc.second = Integer.MIN_VALUE;
    return acc;
  }

  public void accumulate(Top2Accumulator acc, Integer value) {
    if (value > acc.first) {
      acc.second = acc.first;
      acc.first = value;
    } else if (value > acc.second) {
      acc.second = value;
    }
  }

  public void merge(Top2Accumulator acc, Iterable<Top2Accumulator> it) {
    for (Top2Accumulator otherAcc : it) {
      accumulate(acc, otherAcc.first);
      accumulate(acc, otherAcc.second);
    }
  }

  public void emitValue(Top2Accumulator acc, Collector<Tuple2<Integer, Integer>> out) {
    // emit the value and rank
    if (acc.first != Integer.MIN_VALUE) {
      out.collect(Tuple2.of(acc.first, 1));
    }
    if (acc.second != Integer.MIN_VALUE) {
      out.collect(Tuple2.of(acc.second, 2));
    }
  }
}

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

// call function "inline" without registration in Table API
env
  .from("MyTable")
  .groupBy($("myField"))
  .flatAggregate(call(Top2.class, $("value")))
  .select($("myField"), $("f0"), $("f1"));

// call function "inline" without registration in Table API
// but use an alias for a better naming of Tuple2's fields
env
  .from("MyTable")
  .groupBy($("myField"))
  .flatAggregate(call(Top2.class, $("value")).as("value", "rank"))
  .select($("myField"), $("value"), $("rank"));

// register function
env.createTemporarySystemFunction("Top2", Top2.class);

// call registered function in Table API
env
  .from("MyTable")
  .groupBy($("myField"))
  .flatAggregate(call("Top2", $("value")).as("value", "rank"))
  .select($("myField"), $("value"), $("rank"));

import java.lang.Integer
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.table.api._
import org.apache.flink.table.functions.TableAggregateFunction
import org.apache.flink.util.Collector

// mutable accumulator of structured type for the aggregate function
case class Top2Accumulator(
  var first: Integer,
  var second: Integer
)

// function that takes (value INT), stores intermediate results in a structured
// type of Top2Accumulator, and returns the result as a structured type of Tuple2[Integer, Integer]
// for value and rank
class Top2 extends TableAggregateFunction[Tuple2[Integer, Integer], Top2Accumulator] {

  override def createAccumulator(): Top2Accumulator = {
    Top2Accumulator(
      Integer.MIN_VALUE,
      Integer.MIN_VALUE
    )
  }

  def accumulate(acc: Top2Accumulator, value: Integer): Unit = {
    if (value > acc.first) {
      acc.second = acc.first
      acc.first = value
    } else if (value > acc.second) {
      acc.second = value
    }
  }

  def merge(acc: Top2Accumulator, it: java.lang.Iterable[Top2Accumulator]) {
    val iter = it.iterator()
    while (iter.hasNext) {
      val otherAcc = iter.next()
      accumulate(acc, otherAcc.first)
      accumulate(acc, otherAcc.second)
    }
  }

  def emitValue(acc: Top2Accumulator, out: Collector[Tuple2[Integer, Integer]]): Unit = {
    // emit the value and rank
    if (acc.first != Integer.MIN_VALUE) {
      out.collect(Tuple2.of(acc.first, 1))
    }
    if (acc.second != Integer.MIN_VALUE) {
      out.collect(Tuple2.of(acc.second, 2))
    }
  }
}

val env = TableEnvironment.create(...)

// call function "inline" without registration in Table API
env
  .from("MyTable")
  .groupBy($"myField")
  .flatAggregate(call(classOf[Top2], $"value"))
  .select($"myField", $"f0", $"f1")

// call function "inline" without registration in Table API
// but use an alias for a better naming of Tuple2's fields
env
  .from("MyTable")
  .groupBy($"myField")
  .flatAggregate(call(classOf[Top2], $"value").as("value", "rank"))
  .select($"myField", $"value", $"rank")

// register function
env.createTemporarySystemFunction("Top2", classOf[Top2])

// call registered function in Table API
env
  .from("MyTable")
  .groupBy($"myField")
  .flatAggregate(call("Top2", $"value").as("value", "rank"))
  .select($"myField", $"value", $"rank")

우리 Top2 클래스의 accumulate(...) 메서드는 두 개의 입력을 받습니다. 첫 번째는 축적자이고 두 번째는 사용자 정의 입력입니다. 결과를 계산하려면 축적자가 지금까지 축적된 모든 데이터 중 가장 높은 값 2개를 저장해야 합니다. 축적자는 Flink의 체크포인트 메커니즘에 의해 자동으로 관리되며, 정확히 한 번(exactly-once) 의미를 보장하기 위해 실패 시 복원됩니다. 결과 값은 순위 인덱스와 함께 방출됩니다.

필수 및 선택 메서드 (Mandatory and Optional Methods)

각 TableAggregateFunction에 대해 다음 메서드는 필수입니다.

  • createAccumulator()
  • accumulate(...)
  • emitValue(...) 또는 emitUpdateWithRetract(...)

추가로 선택적으로 구현할 수 있는 몇 가지 메서드가 있습니다. 일부 메서드는 시스템이 더 효율적인 쿼리 실행을 허용하고, 다른 일부는 특정 use-case에 필수입니다. 예를 들어 merge(...) 메서드는 집계 함수가 세션 그룹 윈도우 컨텍스트에서 적용되어야 하는 경우 필수입니다(두 세션 윈도우를 "연결"하는 행이 관찰될 때 두 세션 윈도우의 축적자를 결합해야 하기 때문).

TableAggregateFunction의 다음 메서드는 use-case에 따라 필요합니다.

  • retract(...)는 OVER 윈도우에서의 집계에 필요합니다.
  • merge(...)는 많은 유한 집계와 무한 세션 및 홉 윈도우 집계에 필요합니다.
  • emitValue(...)는 유한 및 윈도우 집계에 필요합니다.

TableAggregateFunction의 다음 메서드는 스트리밍 작업의 성능을 개선하는 데 사용됩니다.

  • emitUpdateWithRetract(...)는 retract 모드에서 업데이트된 값을 방출하는 데 사용됩니다.

emitValue(...) 메서드는 항상 축적자에 따라 전체 데이터를 방출합니다. 무한 시나리오에서는 이것이 성능 문제를 일으킬 수 있습니다. Top N 함수를 예로 들어보겠습니다. emitValue(...)는 매번 모든 N 값을 방출합니다. 성능을 개선하기 위해 retract 모드에서 증분적으로 데이터를 출력하는 emitUpdateWithRetract(...)를 구현할 수 있습니다. 즉 업데이트가 있으면 새 업데이트된 레코드를 보내기 전에 이전 레코드를 철회(retract)할 수 있습니다. 이 메서드는 emitValue(...) 메서드보다 우선 사용됩니다.

테이블 집계 함수가 OVER 윈도우에서만 적용될 수 있다면 getRequirements()에서 FunctionRequirement.OVER_WINDOW_ONLY 요구 사항을 반환해 선언할 수 있습니다.

축적자가 많은 양의 데이터를 저장해야 한다면 org.apache.flink.table.api.dataview.ListView와 org.apache.flink.table.api.dataview.MapView가 무한 데이터 시나리오에서 Flink의 상태 백엔드를 활용하는 고급 기능을 제공합니다. 이 고급 기능에 대한 자세한 내용은 해당 클래스 문서를 참조하세요.

일부 메서드는 선택적이거나 오버로드될 수 있으므로 메서드는 생성된 코드에 의해 호출됩니다. 기본 클래스는 항상 구체적 구현 클래스가 오버라이드할 시그니처를 제공하지는 않습니다. 그럼에도 불구하고 언급된 모든 메서드는 public으로 선언되고 static이 아니며, 호출되려면 위에서 언급한 이름과 정확히 일치해야 합니다.

TableAggregateFunction에서 선언되지 않고 생성된 코드에 의해 호출되는 모든 메서드에 대한 자세한 문서는 아래에 나와 있습니다.

accumulate(...)

/*
 * Processes the input values and updates the provided accumulator instance. The method
 * accumulate can be overloaded with different custom types and arguments. An aggregate function
 * requires at least one accumulate() method.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: [user defined inputs] the input value (usually obtained from new arrived data).
 */
public void accumulate(ACC accumulator, [user defined inputs])

retract(...)

/*
 * Retracts the input values from the accumulator instance. The current design assumes the
 * inputs are the values that have been previously accumulated. The method retract can be
 * overloaded with different custom types and arguments. This method must be implemented for
 * bounded OVER aggregates over unbounded tables.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: [user defined inputs] the input value (usually obtained from new arrived data).
 */
public void retract(ACC accumulator, [user defined inputs])

merge(...)

/*
 * Merges a group of accumulator instances into one accumulator instance. This method must be
 * implemented for unbounded session window grouping aggregates and bounded grouping aggregates.
 *
 * param: accumulator the accumulator which will keep the merged aggregate results. It should
 *                    be noted that the accumulator may contain the previous aggregated
 *                    results. Therefore user should not replace or clean this instance in the
 *                    custom merge method.
 * param: iterable    an java.lang.Iterable pointed to a group of accumulators that will be
 *                    merged.
 */
public void merge(ACC accumulator, java.lang.Iterable<ACC> iterable)

emitValue(...)

/*
 * Called every time when an aggregation result should be materialized. The returned value could
 * be either an early and incomplete result (periodically emitted as data arrives) or the final
 * result of the aggregation.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: out                   the collector used to output data.
 */
public void emitValue(ACC accumulator, org.apache.flink.util.Collector<T> out)

emitUpdateWithRetract(...)

/*
 * Called every time when an aggregation result should be materialized. The returned value could
 * be either an early and incomplete result (periodically emitted as data arrives) or the final
 * result of the aggregation.
 *
 * Compared to emitValue(), emitUpdateWithRetract() is used to emit values that have been updated. This method
 * outputs data incrementally in retraction mode (also known as "update before" and "update after"). Once
 * there is an update, we have to retract old records before sending new updated ones. The emitUpdateWithRetract()
 * method will be used in preference to the emitValue() method if both methods are defined in the table aggregate
 * function, because the method is treated to be more efficient than emitValue as it can output
 * values incrementally.
 *
 * param: accumulator           the accumulator which contains the current aggregated results
 * param: out                   the retractable collector used to output data. Use the collect() method
 *                              to output(add) records and use retract method to retract(delete)
 *                              records.
 */
public void emitUpdateWithRetract(ACC accumulator, RetractableCollector<T> out)

철회 예제 (Retraction Example)

다음 예제는 증분 업데이트만 방출하기 위해 emitUpdateWithRetract(...) 메서드를 사용하는 방법을 보여줍니다. 이를 위해 축적자는 이전 및 새로운 상위 2개 값을 모두 유지합니다.

주의: emitUpdateWithRetract 내에서 축적자를 업데이트하지 마세요. function#emitUpdateWithRetract가 호출된 후 GroupTableAggFunction은 function#getAccumulators를 다시 호출해 최신 축적자를 상태로 업데이트하지 않기 때문입니다.

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.functions.TableAggregateFunction;

public static class Top2WithRetractAccumulator {
  public Integer first;
  public Integer second;
  public Integer oldFirst;
  public Integer oldSecond;
}

public static class Top2WithRetract
    extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2WithRetractAccumulator> {

  @Override
  public Top2WithRetractAccumulator createAccumulator() {
    Top2WithRetractAccumulator acc = new Top2WithRetractAccumulator();
    acc.first = Integer.MIN_VALUE;
    acc.second = Integer.MIN_VALUE;
    acc.oldFirst = Integer.MIN_VALUE;
    acc.oldSecond = Integer.MIN_VALUE;
    return acc;
  }

  public void accumulate(Top2WithRetractAccumulator acc, Integer v) {
    acc.oldFirst = acc.first;
    acc.oldSecond = acc.second;
    if (v > acc.first) {
      acc.second = acc.first;
      acc.first = v;
    } else if (v > acc.second) {
      acc.second = v;
    }
  }

  public void emitUpdateWithRetract(
      Top2WithRetractAccumulator acc,
      RetractableCollector<Tuple2<Integer, Integer>> out) {
    if (!acc.first.equals(acc.oldFirst)) {
      // if there is an update, retract the old value then emit a new value
      if (acc.oldFirst != Integer.MIN_VALUE) {
          out.retract(Tuple2.of(acc.oldFirst, 1));
      }
      out.collect(Tuple2.of(acc.first, 1));
    }
    if (!acc.second.equals(acc.oldSecond)) {
      // if there is an update, retract the old value then emit a new value
      if (acc.oldSecond != Integer.MIN_VALUE) {
          out.retract(Tuple2.of(acc.oldSecond, 2));
      }
      out.collect(Tuple2.of(acc.second, 2));
    }
  }
}

import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.table.functions.TableAggregateFunction
import org.apache.flink.table.functions.TableAggregateFunction.RetractableCollector

case class Top2WithRetractAccumulator(
  var first: Integer,
  var second: Integer,
  var oldFirst: Integer,
  var oldSecond: Integer
)

class Top2WithRetract
    extends TableAggregateFunction[Tuple2[Integer, Integer], Top2WithRetractAccumulator] {

  override def createAccumulator(): Top2WithRetractAccumulator = {
    Top2WithRetractAccumulator(
      Integer.MIN_VALUE,
      Integer.MIN_VALUE,
      Integer.MIN_VALUE,
      Integer.MIN_VALUE
    )
  }

  def accumulate(acc: Top2WithRetractAccumulator, value: Integer): Unit = {
    acc.oldFirst = acc.first
    acc.oldSecond = acc.second
    if (value > acc.first) {
      acc.second = acc.first
      acc.first = value
    } else if (value > acc.second) {
      acc.second = value
    }
  }

  def emitUpdateWithRetract(
      acc: Top2WithRetractAccumulator,
      out: RetractableCollector[Tuple2[Integer, Integer]])
    : Unit = {
    if (!acc.first.equals(acc.oldFirst)) {
      // if there is an update, retract the old value then emit a new value
      if (acc.oldFirst != Integer.MIN_VALUE) {
          out.retract(Tuple2.of(acc.oldFirst, 1))
      }
      out.collect(Tuple2.of(acc.first, 1))
    }
    if (!acc.second.equals(acc.oldSecond)) {
      // if there is an update, retract the old value then emit a new value
      if (acc.oldSecond != Integer.MIN_VALUE) {
          out.retract(Tuple2.of(acc.oldSecond, 2))
      }
      out.collect(Tuple2.of(acc.second, 2))
    }
  }
}

프로세스 테이블 함수 (Process Table Functions)

프로세스 테이블 함수(PTF, Process Table Functions)는 Flink SQL과 Table API에서 가장 강력한 함수 종류입니다. 이들은 내장 연산자만큼 기능이 풍부한 사용자 정의 연산자를 구현할 수 있게 해줍니다. PTF는 (파티셔닝된) 테이블을 받아 새로운 테이블을 만들 수 있습니다. Flink의 관리형 상태(managed state), 이벤트 시간(event-time)과 타이머 서비스, 그리고 기본 테이블 변경 로그(changelog)에 접근할 수 있습니다.

개념적으로 PTF는 다른 모든 사용자 정의 함수의 상위 집합입니다. 0개, 1개 또는 여러 개의 테이블을 0개, 1개 또는 여러 개의 행(또는 구조화된 타입)으로 매핑합니다. 스칼라 인자가 지원됩니다. 상태를 갖는 특성 때문에 집계 동작을 구현하는 것도 가능합니다.

PTF는 다음 작업을 활성화합니다.

  • 테이블의 각 행에 변환을 적용합니다.
  • 테이블을 논리적으로 서로 다른 집합으로 파티셔닝하고 집합별로 변환을 적용합니다.
  • 반복 접근을 위해 본 이벤트(seen events)를 저장합니다.
  • 대기, 동기화, 타임아웃을 활성화하는 더 나중 시간에 처리를 계속합니다.
  • 복잡한 상태 머신 또는 규칙 기반 조건부 로직을 사용해 이벤트를 버퍼링하고 집계합니다.

자세한 내용은 PTF 전용 페이지를 참조하세요.

더 알아보기 (Learn more)