Flink1.18 如何配置算子级别的TTL

2024-03-22 21:36

本文主要是介绍Flink1.18 如何配置算子级别的TTL,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1. 解释

从 Flink 1.18 开始,Table API & SQL 支持配置细粒度的状态 TTL 来优化状态使用,可配置粒度为每个状态算子的入边数。具体而言,OneInputStreamOperator 可以配置一个状态的 TTL,而 TwoInputStreamOperator(例如双流 join)则可以分别为左状态和右状态配置 TTL。更一般地,对于具有 K 个输入的 MultipleInputStreamOperator,可以配置 K 个状态 TTL。

2. 使用场景

2.1 为 双流 Join 的左右流配置不同 TTL。

双流 Join 会生成拥有两条输入边TwoInputStreamOperator的状态算子,它用到了两个状态,分别来保存来自左流和右流的更新

2.2 在同一个作业中为不同的状态计算设置不同 TTL

举例来说,假设一个 ETL 作业使用 ROW_NUMBER 进行去重操作后, 紧接着使用 GROUP BY 语句进行聚合操作。 该作业会分别生成两个拥有单条输入边的 OneInputStreamOperator 状态算子。您可以为去重算子和聚合算子的状态分别设置不同的 TTL

​2.3 注意

由于基于窗口的操作(例如窗口连接、窗口聚合、窗口 Top-N 等)和 Interval Join 不依赖于 table.exec.state.ttl 来控制状态保留,因此它们的状态无法在算子级别进行配置

Interval Join 是一种流处理操作,它在两个数据流之间基于时间间隔进行连接这种连接是基于时间窗口的,而不是基于状态的保留时间。在 Interval Join 中,系统会根据两个数据流中元素的时间戳和定义的间隔来决定哪些元素应该被连接在一起

3.如何使用

3.1 生成 Compiled Plan

注意使用时的类型是insert,如果不是insert会报错 Unsupported SQL query! compilePlanSql() only accepts a single SQL statement of type INSERT

CompiledPlan compiledPlan = tableEnv.compilePlanSql("INSERT INTO enriched_orders \n" + "SELECT a.order_id, a.order_line_id, b.order_status, ... \n" + "FROM orders a JOIN line_orders b ON a.order_line_id = b.order_line_id");compiledPlan.writeToFile("/path/to/plan.json");在Flink Sql中使用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;

3.2 修改 Compiled Plan

每个状态算子会显式地生成一个名为 “state” 的 JSON 数组,具有如下结构。 理论上一个拥有 k 路输入的状态算子拥有 k 个状态"state": [{"index": 0,"ttl": "0 ms","name": "${1st input state name}"},{"index": 1,"ttl": "0 ms","name": "${2nd input state name}"},...]找到您需要修改的状态算子,将 TTL 的值设置为一个正整数,注意需要带上时间单位毫秒。举例来说,如果想将当前状态算子的 TTL 设置为 1 小时,您可以按照如下格式修改 JSON:{"index": 0,"ttl": "3600000 ms","name": "${1st input state name}"
}理论上,下游状态算子的 TTL 不应小于上游状态算子的 TTL。
保存好文件,然后使用 EXECUTE PLAN 语句来提交作业

3.3 执行Compiled Plan

tableEnv.loadPlan(PlanReference.fromFile("/path/to/plan.json")).execute().await();在Flink Sql中使用EXECUTE PLAN '/path/to/plan.json'

这篇关于Flink1.18 如何配置算子级别的TTL的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/836244

相关文章

最新版IDEA配置 Tomcat的详细过程

《最新版IDEA配置Tomcat的详细过程》本文介绍如何在IDEA中配置Tomcat服务器,并创建Web项目,首先检查Tomcat是否安装完成,然后在IDEA中创建Web项目并添加Web结构,接着,... 目录配置tomcat第一步,先给项目添加Web结构查看端口号配置tomcat    先检查自己的to

Servlet中配置和使用过滤器的步骤记录

《Servlet中配置和使用过滤器的步骤记录》:本文主要介绍在Servlet中配置和使用过滤器的方法,包括创建过滤器类、配置过滤器以及在Web应用中使用过滤器等步骤,文中通过代码介绍的非常详细,需... 目录创建过滤器类配置过滤器使用过滤器总结在Servlet中配置和使用过滤器主要包括创建过滤器类、配置过滤

在 VSCode 中配置 C++ 开发环境的详细教程

《在VSCode中配置C++开发环境的详细教程》本文详细介绍了如何在VisualStudioCode(VSCode)中配置C++开发环境,包括安装必要的工具、配置编译器、设置调试环境等步骤,通... 目录如何在 VSCode 中配置 C++ 开发环境:详细教程1. 什么是 VSCode?2. 安装 VSCo

在Spring中配置Quartz的三种方式

《在Spring中配置Quartz的三种方式》SpringQuartz是一个任务调度框架,它允许我们定期执行特定的任务,在Spring中,我们可以通过多种方式来配置Quartz,包括使用​​@Sche... 目录介绍使用 ​​@Scheduled​​ 注解XML 配置Java 配置1. 创建Quartz配置

Kibana的安装和配置全过程

《Kibana的安装和配置全过程》Kibana是一个开源的数据分析和可视化平台,它与Elasticsearch紧密集成,提供了一个直观的Web界面,使您可以快速地搜索、分析和可视化数据,在本文中,我们... 目录Kibana的安装和配置1.安装Java运行环境2.下载Kibana3.解压缩Kibana4.配

tomcat在nginx中的配置方式

《tomcat在nginx中的配置方式》文章介绍了如何在Linux系统上安装和配置Tomcat,并通过Nginx进行代理,首先,下载并解压Tomcat压缩包,然后启动Tomcat并查看日志,接着,配置... 目录一、下载安装tomcat二、启动tomcat三、配置nginx总结提示:文章写完后,目录可以自动

Zookeeper安装和配置说明

一、Zookeeper的搭建方式 Zookeeper安装方式有三种,单机模式和集群模式以及伪集群模式。 ■ 单机模式:Zookeeper只运行在一台服务器上,适合测试环境; ■ 伪集群模式:就是在一台物理机上运行多个Zookeeper 实例; ■ 集群模式:Zookeeper运行于一个集群上,适合生产环境,这个计算机集群被称为一个“集合体”(ensemble) Zookeeper通过复制来实现

CentOS7安装配置mysql5.7 tar免安装版

一、CentOS7.4系统自带mariadb # 查看系统自带的Mariadb[root@localhost~]# rpm -qa|grep mariadbmariadb-libs-5.5.44-2.el7.centos.x86_64# 卸载系统自带的Mariadb[root@localhost ~]# rpm -e --nodeps mariadb-libs-5.5.44-2.el7

hadoop开启回收站配置

开启回收站功能,可以将删除的文件在不超时的情况下,恢复原数据,起到防止误删除、备份等作用。 开启回收站功能参数说明 (1)默认值fs.trash.interval = 0,0表示禁用回收站;其他值表示设置文件的存活时间。 (2)默认值fs.trash.checkpoint.interval = 0,检查回收站的间隔时间。如果该值为0,则该值设置和fs.trash.interval的参数值相等。

NameNode内存生产配置

Hadoop2.x 系列,配置 NameNode 内存 NameNode 内存默认 2000m ,如果服务器内存 4G , NameNode 内存可以配置 3g 。在 hadoop-env.sh 文件中配置如下。 HADOOP_NAMENODE_OPTS=-Xmx3072m Hadoop3.x 系列,配置 Nam