介绍

转换算子是无法访问事件的 时间戳 信息和 水位线 信息的。在一些应用场景下，需要用到这些信息

例如：MapFunction 这样的 map 转换算子就无法访问时间戳或者当前事件的事件时间。

所以，DataStream API提供了一系列的 Low-Level 函数。可以访问：

有 open()、close() 方法，具备生命周期
event：数据流中的每个元素
state：状态
timers：（事件时间或处理时间）定时器，仅在keyedStream中可以访问。
watermark
注册使用定时器
等等

例如：超时事件，订单超时取消

Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑(使用window函数和转换算子无法实现)。例如，Flink SQL就是使用Process Function实现的。

使用场景

复杂状态管理：结合状态存储中间结果（如累计计数、缓存关联数据）。
基于时间的业务：如会话窗口（Session Window）、延迟数据处理、超时告警。
数据分流：通过侧输出流分离正常数据和异常数据。
自定义窗口逻辑：实现 Flink 内置窗口不支持的特殊窗口逻辑。
需要连接数据库、关闭数据库

通用方法

open(Configuration parameters)：初始化方法，可用于创建资源（如数据库连接、状态初始化）
close()：清理方法，用于释放资源（如关闭连接）

ProcessFunction 是 Flink 中处理流数据的底层核心接口，提供了对事件时间、状态、定时器等底层特性的直接访问，适用于复杂业务逻辑（如实时监控、超时告警、动态状态管理等）。其生命周期由 Flink 框架管理，核心方法（open()、processElement()、onTimer()、close() 等）的执行时机有明确规则。

open(Configuration)

执行时机：
在 ProcessFunction 实例被创建并初始化后，首次处理数据之前执行（仅执行一次）。
具体来说，当 Flink 作业启动、算子（Operator）初始化完成后，会调用该方法。
作用：
- 初始化资源（如数据库连接、Redis 客户端、Kafka 生产者等）。
- 初始化状态（通过 getRuntimeContext().getState(...) 创建 ValueState、ListState 等）。
- 读取配置参数（通过 parameters 获取配置项）。

例子：

  @Override
  public void open(Configuration parameters) throws Exception {
      super.open(parameters);
      // 初始化数据库连接
      dbConnection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "user", "pass");
      // 初始化状态
      countState = getRuntimeContext().getState(new ValueStateDescriptor<>("count", Integer.class));
  }

close()

执行时机：
在 ProcessFunction 实例生命周期结束时执行（仅执行一次）。
触发场景包括：作业正常停止、作业失败重启、Checkpoint 时的实例替换等。
作用：
- 释放资源（如关闭数据库连接、Redis 客户端、文件流等），避免资源泄漏。
- 执行最终清理逻辑（如提交未完成的事务）。

例子：

  @Override
  public void close() throws Exception {
      super.close();
      // 关闭数据库连接
      if (dbConnection != null && !dbConnection.isClosed()) {
          dbConnection.close();
      }
  }

生命周期

一个 ProcessFunction 实例的完整生命周期如下：

算子初始化 → open() → [processElement()类似方法 * N次（处理每个元素）] → [onTimer() * M次（定时器触发）] → close() → 实例销毁

并行度影响

Flink 算子按并行度（Parallelism）创建多个实例，每个并行实例都有独立的生命周期，即每个实例都会单独执行 open()、close()，且状态和定时器在实例间隔离。

关键注意事项

资源初始化与释放：
- 耗时资源（如连接池）必须在 open() 中初始化，避免在 processElement() 中重复创建（否则严重影响性能）。
- 资源必须在 close() 中释放，否则可能导致作业重启时资源泄露（如端口占用、连接数耗尽）。
状态与序列化：
- 状态（ValueState 等）必须在 open() 中初始化，且状态变量需用 transient 标记（避免 Checkpoint 时不必要的序列化）。
- 非序列化资源（如数据库连接）不能作为状态存储，需在 open() 中重新初始化（Checkpoint 恢复时会重新调用 open()）。
异常处理：
- open() 中若初始化失败（如连接数据库超时），会导致算子启动失败，作业直接报错。
- processElement() 中抛出的异常会导致当前元素处理失败，需通过 try-catch 捕获并处理（如发送到侧输出流）。

总结

ProcessFunction 的方法执行时机严格遵循生命周期：

open()：初始化阶段，准备资源和状态；
processElement() 类似方法：数据处理阶段，每个元素触发一次；
onTimer()：时间触发阶段，定时器到期时执行；
close()：销毁阶段，释放资源。

参考：
https://www.jianshu.com/p/ca67629d8297
https://blog.csdn.net/weixin_47491957/article/details/124451027

原文出处：/show_1IX4FBFCaqEo.html

flink1.12.x Process Function处理函数介绍、分类、通用方法

介绍