自定义处理表值函数（UDPTF）

定义行语义和集合语义的表参数

定义了行语义表参数 ArgumentTrait.ROW_SEMANTIC_TABLE 的 PTF 假定各行之间不存在关联，因此每一行都可以独立处理。框架可以自由决定如何将当前行分配到某个 PTF 实例处理。

定义了集合语义表参数 ArgumentTrait.SET_SEMANTIC_TABLE 的 PTF 假定各行之间存在关联，PARTITION BY 子句定义用于关联的列，框架将会把数据按照分区键进行 shuffle，确保属于同一分区键的所有行将会被分配到同一个并发上处理。也可以通过声明参数 ArgumentTrait.OPTIONAL_PARTITION_BY 不定义分区键，此时所有数据将会被同一个 PTF 实例处理。

例如：

// 采用了行语义的 UDPTF
// 使用方式为 SELECT * FROM f(r => TABLE t)
public void eval(
  @ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) {
  ...
}

// 采用了集合语义的 UDPTF
// 使用方式为 SELECT * FROM f(r => TABLE t PARTITION BY name)
//        或 SELECT * FROM f(r => TABLE t)
public void eval(
  @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row input) {
  ...
}

详情请参考 表语义和虚拟处理器。

访问状态

一个被定义成集合语义的 PTF 可以是有状态的。框架将会在故障或者重启期间存储和恢复这些状态。状态按照 PARTITION BY 子句定义的分区键分区，如果未定义分区键，则为单个分区（即为 Flink 中的 keyed state）。

状态可以在 PTF 中用 StateHint定义，并且可以显示声明状态类型、状态过期时间等参数。例如：

class StateFunction extends ProcessTableFunction<String> {
  /** POJO for state. All fields nullable. */
  public static class Score {
    public String s;
    public Integer i;
  }

  public void eval(
    Context ctx,
    // 显式声明该状态的数据结构
    @StateHint(type = @DataTypeHint("ROW<emitted BOOLEAN>")) Row s0,
    // 显式声明该状态过期时间为 5 天
    @StateHint(ttl = "5 days") Score s1,
    // 显式声明该状态过期时间为 0 毫秒（即永不过期），无时间单位即为毫秒
    @StateHint(ttl = "0") Score s2,
    // 该状态过期时间为参数 table.exec.state.ttl 的值
    @StateHint Score s3,
    @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row r) {
    ...
    // 可以通过 ctx.clearState("s1") 或 ctx.clearAllState() 
    // 来清理状态，防止状态过大
  }
}

目前 PTF 内可以使用以下 3 种状态类型

• Value State：表示单个值。是 PTF 中的默认状态类型。

• List State：表示值列表，可以进行追加、删除、迭代操作。

• Map State：表示一个键值对的映射，用于高效查询、修改和删除某个条目。

默认情况下， PTF 中的状态为 Value State，每次访问都需要从状态后端读取和写入完整的数据。为了优化状态访问和不必要的序列化和反序列化，状态对象可以声明为：

• org.apache.flink.table.api.dataview.ListView （即为 List State）

• org.apache.flink.table.api.dataview.MapView （即为 Map State）

详情请参考状态。

时间和定时器

PTF 原生支持事件时间。您可以借助 PTF 中的事件时间和定时器，自定义处理延迟触发、迟到数据处理等业务逻辑。

时间

每个 PTF 在调用时都有一个可选的 on_time 来声明被定义成事件时间的时间属性列，也可以通过声明参数 ArgumentTrait.REQUIRE_ON_TIME 来强制要求调用时必须声明 on_time 列。

SELECT * FROM f(..., on_time => DESCRIPTOR(`my_timestamp`));

您可以通过 Context#timeContext(Class) 获取 TimeContext 来访问时间进行自定义操作。该时间上下文的作用域始终限定于当前正在处理的事件。该事件可以是当前输入行， 也可以是正在触发的定时器。时间和定时器的时间戳可以表示为 java.time.Instant、java.time.LocalDateTime 或 Long。 这些时间戳以纪元以来的毫秒数为基础，不考虑本地会话时区。可以将时间类作为参数传给 timeContext()。

/** 为给定键发送 ping 的函数。在观察到该键的最后一个事件一分钟后发送 ping。 */
public static class PingLaterFunction
  extends ProcessTableFunction<String> {
    public void eval(
      Context ctx,
      @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, 
                     ArgumentTrait.REQUIRE_ON_TIME}) 
      Row input
    ) {
      TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);
      // 替换现有定时器，因此必要时可能会重新开始一分钟倒计时
      timeCtx.registerOnTime("ping", timeCtx.time().plus(Duration.ofMinutes(1)));
    }

    public void onTimer(OnTimerContext onTimerCtx) {
      collect("ping");
    }
  }

TimeContext 提供了多个函数，可以获取到当前处理数据的事件时间、触发定时器的时间、当前输入侧的水印时间、当前 PTF 实例的水印时间。详情请参考时间。

定时器

集合语义的 PTF 支持使用定时器。定时器允许在稍后的时间点处理数据，从而实现等待、同步或超时等操作。当水位线推进时，定时器会在注册的时间被触发。

需要使用定时器的 PTF 必须声明 onTimer() 方法，以处理定时器事件。 onTimer() 的方法签名如下：

onTimer( <on timer context>? , <state entry>* )

框架会在故障或重启期间存储和恢复定时器。因此，定时器是一种特殊的状态。 同样，定时器的作用域限定于 PARTITION BY 子句定义的状态分区中。定时器只能在当前分区键中注册和删除。当定时器过多影响性能时，您可以通过 Context#clearAllTimers() 或 TimeContext#clearTimer(String) 清理不再需要的定时器。详情请参考定时器。

排序

集合语义的 PTF 可以选择在调用时指定 ORDER BY 子句，以定义每个分区内各行的处理顺序。ORDER BY 子句保证各行按指定顺序传递给 eval() 方法。

ORDER BY 子句要求第一列是时间属性列（即已声明水印的 TIMESTAMP 或 TIMESTAMP_LTZ 列）。第一个 ORDER BY 列必须按升序指定，以确保按事件时间顺序处理各行。 还可以指定其他列作为次级排序键，以定义具有相同时间戳的各行的顺序。详情请参考排序。

SELECT * FROM my_ptf(
  input_table => TABLE source_table
    PARTITION BY user_id
    ORDER BY (event_time ASC, priority DESC NULLS FIRST)
)

多表处理

PTF 可以同时处理多张表。eval() 方法可以指定多个表参数以支持多个输入。所有表参数都必须声明 为集合语义并采用一致的分区方式，即所有相关表参数的 PARTITION BY 子句中的列数及其数据类型必须一致。

来自任一输入的行每次只向函数传递一行。因此，同一时刻只有一个表参数非 null。使用 null 检查来确定当前正在处理哪个输入。

以下是自定义连接的一个例子：

/** 向客户问候，并在有可用信息时推荐上次所购物品的函数。 */
public static class GreetingWithLastPurchase extends ProcessTableFunction<String> {

  /** 在状态中保留最近购买的物品。 */
  public static class LastItemState {
    public String lastItem;
  }

  /** eval() 方法接收两个 @ArgumentHint(SET_SEMANTIC_TABLE) 参数 */
  public void eval(
    @StateHint LastItemState state,
    @ArgumentHint(SET_SEMANTIC_TABLE) Row visit,
    @ArgumentHint(SET_SEMANTIC_TABLE) Row purchase) {

    // 处理来自 Purchases 表的行
    if (purchase != null) {
      state.lastItem = purchase.getFieldAs("item");
    }

      // 处理来自 Visits 表的行
    else if (visit != null) {
      if (state.lastItem == null) {
        collect("Hello " + visit.getFieldAs("name") + ", let me know if I can help!");
      } else {
        collect("Hello " + visit.getFieldAs("name") + ", here to buy " + state.lastItem + " again?");
      }
    }
  }
}

详情请参考多表。

透传列

根据表语义以及是否定义了 on_time 参数，框架会为 PTF 函数的输出添加附加列。

对于集合语义的 PTF，输出会以PARTITION BY 列为前缀。对于带有 on_time 参数的调用，输出会以 <rowtime> 为后缀。因此，默认的输出格式为：

<PARTITION BY keys> | <function output> | <rowtime>

您可以定义 ArgumentTrait.PASS_COLUMNS_THROUGH 参数，来让 PTF 输出所有的输入表的列。例如有一张输入表 t（包含列 k 和 v），和一个 PTF 函数 f()（产生列 c1 和 c2），查询 SELECT * FROM f(table_arg => TABLE t PARTITION BY k) 的输出类型为：

默认： 
    | k | c1 | c2 |
定义 ArgumentTrait.PASS_COLUMNS_THROUGH 后： 
    | k | v | c1 | c2 |

详情请参考透传列。

更新和变更日志

默认情况下，PTF 仅支持消费、产生仅追加流。下面这些参数或接口，可以让 PTF 处理更新流数据

以下是 PTF 消费更新流的示例：

@DataTypeHint("ROW<flag STRING, sum INT>")
public static class ToChangelogFunction extends ProcessTableFunction<Row> {
  public void eval(@ArgumentHint({SET_SEMANTIC_TABLE, SUPPORT_UPDATES}) Row input) {
    // 转发 sum 列，并将行类型作为字符串列包含进来。
    Row changelogRow =
      Row.of(
        input.getKind().toString(),
        input.getField("sum"));

    collect(changelogRow);
  }
}

以下是 PTF 产生更新流的示例：

@DataTypeHint("ROW<sum INT>")
public static class CustomAggregation
  extends ProcessTableFunction<Row>
  implements ChangelogFunction {

    @Override
    public ChangelogMode getChangelogMode(ChangelogContext changelogContext) {
      // 告知框架 PTF 生成以回撤流
      return ChangelogMode.all();
    }

    public static class Accumulator {
      public Integer sum = 0;
    }

    public void eval(
      @StateHint Accumulator state,
      @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
      int score = input.getFieldAs("score");

      // 负数状态表示该分区
      // 键此前已被标记为无效
      if (state.sum == -1) {
        return;
      }

      // 负分数会将整个聚合结果标记为无效。
      if (score < 0) {
        // 为受影响的分区键发出 -D，并
        // 在状态中标记失效。所有后续操作
        // 和接收器都将删除该聚合结果。
        collect(Row.ofKind(RowKind.DELETE, state.sum));
        state.sum = -1;
      } else {
        if (state.sum == 0) {
          // 为第一个有效聚合结果发出 +I。
          state.sum += score;
          collect(Row.ofKind(RowKind.INSERT, state.sum));
        } else {
          // 发出 -U（包含旧聚合结果）和 +U（包含新聚合结果）
          // 以编码更新。
          collect(Row.ofKind(RowKind.UPDATE_BEFORE, state.sum));
          state.sum += score;
          collect(Row.ofKind(RowKind.UPDATE_AFTER, state.sum));
        }
      }
    }
  }

详情请参考更新和变更日志。