官宣｜Apache Flink 1.19 发布公告

Apache Flink PMC（项目管理委员）很高兴地宣布发布 Apache Flink 1.19.0。与往常一样，这是一个充实的版本，包含了广泛的改进和新功能。总共有 162 人为此版本做出了贡献，完成了 33 个 FLIPs、解决了 600 多个问题。感谢各位贡献者的支持！

一、Flink SQL 提升

源表自定义并行度

现在，在 Flink 1.19 中，您可以通过选 scan.parallelism 设置自定义并行度，以调整性能。第一个可用的连接器是 DataGen（ Kafka 连接器即将推出）。下面是一个使用 SQL Client 的示例：

-- set parallelism within the ddl
CREATE TABLE Orders (
    order_number BIGINT,
    price        DECIMAL(32,2),
    buyer        ROW<first_name STRING, last_name STRING>,
    order_time   TIMESTAMP(3)
) WITH (
    'connector' = 'datagen',
    'scan.parallelism' = '4'
);

-- or set parallelism via dynamic table option
SELECT * FROM Orders /*+ OPTIONS('scan.parallelism'='4') */;

更多信息

• 文档
• FLIP-367: Support Setting Parallelism for Table/SQL Sources

可配置的 SQL Gateway Java 选项

一个用于指定 Java 选项的新选项 env.java.opts.sql-gateway ，这样你就可以微调内存设置、垃圾回收行为和其他相关 Java 参数。

更多信息

• FLINK-33203

使用 SQL 提示配置不同的状态 TTL

从 Flink 1.18 开始，Table API 和 SQL 用户可以通过 SQL 编译计划为有状态操作符单独设置状态存续时间 ( TTL )。在 Flink 1.19 中，用户可以使用 STATE_TTL 提示，以更灵活的方式直接在查询中为常规连接和分组聚合指定自定义 TTL 值。

这一改进意味着您不再需要修改编译后的计划，就能为这些常用操作符设置特定的 TTL。引入 STATE_TTL 提示后，您可以简化工作流程，并根据操作要求动态调整 TTL。

下面是一个例子：

-- set state ttl for join
SELECT /*+ STATE_TTL('Orders'= '1d', 'Customers' = '20d') */ *
FROM Orders LEFT OUTER JOIN Customers
    ON Orders.o_custkey = Customers.c_custkey;

-- set state ttl for aggregation
SELECT /*+ STATE_TTL('o' = '1d') */ o_orderkey, SUM(o_totalprice) AS revenue
FROM Orders AS o
GROUP BY o_orderkey;

更多信息

• 文档
• FLIP-373: Support Configuring Different State TTLs using SQL Hint

函数和存储过程支持命名参数

现在，在调用函数或存储过程时可以使用命名参数。使用命名参数时，用户无需严格指定参数位置，只需指定参数名称及其相应值即可。同时，如果没有指定非必要参数，这些参数将默认为空值。

下面是一个使用命名参数定义带有一个必选参数和两个可选参数的函数的示例：

public static class NamedArgumentsTableFunction extends TableFunction<Object> {

    @FunctionHint(
            output = @DataTypeHint("STRING"),
            arguments = {
                    @ArgumentHint(name = "in1", isOptional = false, type = @DataTypeHint("STRING")),
                    @ArgumentHint(name = "in2", isOptional = true, type = @DataTypeHint("STRING")),
                    @ArgumentHint(name = "in3", isOptional = true, type = @DataTypeHint("STRING"))})
    public void eval(String arg1, String arg2, String arg3) {
        collect(arg1 + ", " + arg2 + "," + arg3);
    }

}

在 SQL 中调用函数时，可以通过名称指定参数，例如：

SELECT * FROM TABLE(myFunction(in1 => 'v1', in3 => 'v3', in2 => 'v2'))

可选参数也可以省略：

SELECT * FROM TABLE(myFunction(in1 => 'v1'))

更多信息

• 文档
• FLIP-387: Support named parameters for functions and call procedures

Window TVF 聚合功能

• 支持流模式下的 SESSION Window TVF

现在，用户可以在流模式下使用 SESSION Window TVF。下面是一个简单的示例：

-- session window with partition keys
SELECT * FROM TABLE(
   SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES));

-- apply aggregation on the session windowed table with partition keys
SELECT window_start, window_end, item, SUM(price) AS total_price
FROM TABLE(
    SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
GROUP BY item, window_start, window_end;

• Window TVF 聚合支持处理更新流窗口聚合运算符（基于窗口 TVF 函数生成）现在可以顺利处理更新流（如 CDC 数据源等）。建议用户从传统的 窗口聚合迁移到新语法，以获得更全面的功能支持。

更多信息

• 文档

新的 UDF 类型：AsyncScalarFunction

常见的 UDF 类型 ScalarFunction 可以很好地处理 CPU 密集型操作，但对于 IO 密集型或其他长时间运行的计算则效果不佳。在 Flink 1.19 中，我们新增了 AsyncScalarFunction ，它是一种用户定义的异步 ScalarFunction ，允许异步发出并发函数调用。

更多信息

• FLIP-400: AsyncScalarFunction for asynchronous scalar function support

Regular Join 支持 MiniBatch 优化

消息放大是 Flink 中执行级联连接时的一个痛点，现在在 Flink 1.19 中得到了解决，新的 MiniBatch 优化可用于 Regular Join，以减少此类级联连接场景中的中间结果。

更多信息

• minibatch-regular-joins 文档.
• FLIP-415: Introduce a new join operator to support minibatch

二、Runtime & Coordination 提升

批作业支持源表动态并行度推导

在 Flink 1.19 中，我们支持批作业的源表动态并行度推导，允许源连接器根据实际消耗的数据量动态推断并行度。

与以前的版本相比，这一功能有了重大改进，以前的版本只能为源节点分配固定的默认并行度。

源连接器需要实现推理接口，以启用动态并行度推理。目前，FileSource 连接器已经开发出了这一功能。

此外，配置 execution.batch.adaptive.auto-parallelism.default-source-parallelism 将被用作源并行度推理的上限。现在，它不会默认为 1。取而代之的是，如果没有设置，将使用通过配置 execution.batch.adaptive.auto-parallelism.max-parallelism 设置的允许并行度上限。如果该配置也未设置，则将使用默认的并行度设置 parallelism.default 或 StreamExecutionEnvironment#setParallelism() 。

更多信息

• 文档.
• FLIP-379: Support dynamic source parallelism inference for batch jobs

Flink Configuration 支持标准 YAML 格式

从 Flink 1.19 开始，Flink 正式全面支持标准 YAML 1.2 语法。默认配置文件已改为 config.yaml ，放置在 conf/directory 中。如果用户想使用传统的配置文件 flink-conf.yaml ，只需将该文件复制到 conf/directory 中即可。一旦检测到传统配置文件 flink-conf.yml ，Flink 就会优先使用它作为配置文件。而在即将推出的 Flink 2.0 中， flink-conf.yaml 配置文件将不再起作用。

更多信息

• flink-configuration-file 文档
• FLIP-366: Support standard YAML for Flink configuration

在 Flink Web 上 Profiling JobManager/TaskManager

在 Flink 1.19 中，我们支持在 JobManager/TaskManager 级别触发 Profile，允许用户创建具有任意时间间隔和事件模式（由 async-profiler 支持）的 Profile 实例。用户可以在 Flink Web UI 中轻松提交剖析并导出结果。

例如，用户只需在确定存在性能瓶颈的候选任 JobManager/TaskManager 后，通过 “Create Profiling Instance” 提交一个具有指定周期和模式的 Profile 实例：

Profile 结果：

更多信息

• 文档
• FLIP-375: Built-in cross-platform powerful java profiler

新增管理员 JVM 选项配置选项

有一组管理员 JVM 选项可供使用，它们是用户设置的额外 JVM 选项的前缀，用于全平台范围的 JVM 调整。

更多信息

• 文档
• FLIP-397: Add config options for administrator JVM options

三、Checkpoints 提升

Source 反压时支持使用更大的 Checkpointing 间隔

引入 ProcessingBacklog 的目的是为了说明处理记录时应采用低延迟还是高吞吐量。ProcessingBacklog 可由 Source 算子设置，并可用于在运行时更改作业的检查点间隔。

更多信息

• FLINK-32514
• [FLIP-309: Support using larger checkpointing interval when source is processing backlog]

CheckpointsCleaner 并行清理单个检查点状态

现在，在处置不再需要的检查点时，ioExecutor 会并行处置每个状态句柄/状态文件，从而大大提高了处置单个检查点的速度（对于大型检查点，处置时间可从 10 分钟缩短至 < 1 分钟）。可以通过设置为 false 恢复旧版本的行为。

更多信息

• FLINK-33090

通过命令行客户端触发 Checkpoints

命令行界面支持手动触发检查点。

使用方法：

./bin/flink checkpoint $JOB_ID [-full]

如果指定"-full "选项，就会触发完全检查点。否则，如果作业配置为定期进行增量检查点，则会触发增量检查点。

更多信息

• FLINK-6755

四、Connector API提升

与 Source API 一致的 SinkV2 新接口

在 Flink 1.19 中，SinkV2 API 做了一些修改，以便与 Source API 保持一致。以下接口已被弃用： TwoPhaseCommittingSink、StatefulSink 、WithPreWriteTopology、WithPreCommitTopology、WithPostCommitTopology 。引入了以下新接口 CommitterInitContext 、CommittingSinkWriter 、 WriterInitContext 、StatefulSinkWrite。更改了以下接口方法的参数： Sink#createWriter 。 在 1.19 版本发布期间，原有接口仍将可用，但会在后续版本中移除。

更多信息

• FLINK-33973
• FLIP-372: Enhance and synchronize Sink API to match the Source API

用于跟踪 Committables 状态的新 Committer 指标

修改了 TwoPhaseCommittingSink#createCommitter 方法的参数化，新增了 CommitterInitContext 参数。原来的方法在 1.19 版本发布期间仍然可用，但会在后续版本中移除。

更多信息

• FLINK-25857
• FLIP-371: Provide initialization context for Committer creation in TwoPhaseCommittingSink

五、重要API弃用

为了给 Flink 2.0 版本做准备，社区决定正式废弃多个已接近生命周期终点的 API。

• Flink’s org.apache.flink.api.common.time.Time 现已被正式弃用，并将在 Flink 2.0 中删除。引入了支持 Duration 类的方法，以取代已废弃的基于 Time 的方法。
• org.apache.flink.runtime.jobgraph.RestoreMode#LEGACY 已被弃用。请使用 RestoreMode#CLAIM 或 RestoreMode#NO_CLAIM 模式，以在还原时获得清晰的状态文件所有权。
• 旧的解决模式兼容性的方法已被弃用，请参考迁移说明迁移至新方法： Migrating from deprecated TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer) before Flink 1.19.
• 通过硬代码配置序列化行为已被弃用，例如 ExecutionConfig#enableForceKryo。请使用选 pipeline.serialization-config 、pipeline.force-avr 、pipeline.force-kryo 和 pipeline.generic-types。实例级序列化器的注册已被弃用，请使用类级序列化器。
• 除了 getString(String key, String defaultValue) 和 setString(String key, String value)，我们已废弃所有 setXxx 和 getXxx 方法，如：setInteger 、setLong 、getInteger 和 getLong 等。 建议用户和开发人员使用以 ConfigOption 代替字符串作为键的 get 和 set 方法。
• StreamExecutionEnvironment 、CheckpointConfig 和 ExecutionConfig 中的非 ConfigOption 对象及其相应的 getter/setter 接口现已废弃。这些对象和方法计划在 Flink 2.0 中删除。已废弃的接口包括重启策略（ RestartStrategy ）、检查点存储（ CheckpointStorage ）和状态后端（ StateBackend ）的getter 和 setter 方法。
• org.apache.flink.api.common.functions.RuntimeContext#getExecutionConfig 现已被正式弃用，并将在 Flink 2.0 中删除。请使用 getGlobalJobParameters() 或 isObjectReuseEnabled()。
• org.apache.flink.api.common.functions.RichFunction#open(Configuration parameters) 方法已被弃用，并将在未来版本中删除。我们鼓励用户迁移到新的RichFunction#open(OpenContext openContext)。
• org.apache.flink.configuration.AkkaOptions 已被弃用，取而代之的是 RpcOptions 。

六、升级说明

Apache Flink 社区努力确保升级过程尽可能平稳, 但是升级到 1.19 版本可能需要用户对现有应用程序做出一些调整。请参考 Release Notes 获取更多的升级时需要的改动与可能的问题列表细节。

贡献者列表