Flink1.18新特性生产环境应用的重点解读！

news2024/11/28 12:42:07

大家好，我是你们的群主王知无呀。

Flink 1.18已经于近期发布了。在这个新版本中新增了很多新的功能和特性。在这些特性中，有一些是生产环境非常重要的能力，大家在使用过程中可以重点参考和了解其中的原理。

算子级别状态保留时间TTL设置

首先，在流处理的提升上，从 Flink 1.18 版本开始，Table API 和 SQL 用户可以为有状态的算子单独设置状态保留时间 (TTL)。

这个功能是一个非常实用和重要的功能，在以前的版本中，状态保留时间只能在 pipeline 级别使用配置项通过table.exec.state.ttl进行控制。引入算子级别的状态保留后，用户现在可以根据其具体需求优化资源使用。

现在可以为左侧和右侧流设置不同的 TTL，这有助于大数据量的状态大小控制，可以有效减少状态大小，在失败恢复，重启上线等场景中，任务可以更快恢复。

但是在公开的Flink1.18版本给出的官方用法中，采用了修改JSON File这种不太易用的方式：

-- left source table
CREATE TABLE Orders (
    `order_id` INT,
    `line_order_id` INT
) WITH (
    'connector'='...'
);

-- right source table
CREATE TABLE LineOrders (
    `line_order_id` INT,
    `ship_mode` STRING
) WITH (
    'connector'='...'
);

-- sink table
CREATE TABLE OrdersShipInfo (
    `order_id` INT,
    `line_order_id` INT,
    `ship_mode` STRING
) WITH (
    'connector' = '...'
);

COMPILE PLAN '/path/to/plan.json' FOR
INSERT INTO OrdersShipInfo
SELECT a.order_id, a.line_order_id, b.ship_mode 
FROM Orders a JOIN LineOrders b 
    ON a.line_order_id = b.line_order_id;

然后通过修改这个PLAN中的json数据达到分别设置TTL的目的：

上图中的PLAN中有针对算子的state状态设置，可以根据需要修改左右流的状态：

在FLIP-292中，社区也给出了这么做的理由：

水印对齐和空闲检测

在此之前，这两个功能只能在DatStream API中使用：

-- configure in table options
CREATE TABLE user_actions (
  ...
  user_action_time TIMESTAMP(3),
  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
  'scan.watermark.idle-timeout'='1min',
  ...
);
 
-- use 'OPTIONS' hint
select ... from source_table 
/**
  OPTIONS('scan.watermark.idle-timeout'='1min') 
*/

-- configure in table options
CREATE TABLE user_actions (
...
user_action_time TIMESTAMP(3),
  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
'scan.watermark.alignment.group'='alignment-group-1',
'scan.watermark.alignment.max-drift'='1min',
'scan.watermark.alignment.update-interval'='1s',
...
);
 
-- use 'OPTIONS' hint
select ... from source_table 
/** OPTIONS(
'scan.watermark.alignment.group'='alignment-group-1', 
'scan.watermark.alignment.max-drift'='1min', 
'scan.watermark.alignment.update-interval'='1s') 
*/

其中水印对齐在在多并行度下,Watermark 会在每个并行度的 source 处或者其他算子内部添加，并且需要在进行对齐。

空闲检测就更有用了，我们在很多业务场景中经常会有Source端数据迟迟不来，导致下游某些酸子不能触发计算，在之前我们可以通过设置table.exec.source.idle-timeout全局生效，现在我们可以在不同的源上设置不同的超时时间了。