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

相关文章

VSCode配置Anaconda Python环境的实现

《VSCode配置AnacondaPython环境的实现》VisualStudioCode中可以使用Anaconda环境进行Python开发,本文主要介绍了VSCode配置AnacondaPytho... 目录前言一、安装 Visual Studio Code 和 Anaconda二、创建或激活 conda

IDEA连接达梦数据库的详细配置指南

《IDEA连接达梦数据库的详细配置指南》达梦数据库(DMDatabase)作为国产关系型数据库的代表,广泛应用于企业级系统开发,本文将详细介绍如何在IntelliJIDEA中配置并连接达梦数据库,助力... 目录准备工作1. 下载达梦JDBC驱动配置步骤1. 将驱动添加到IDEA2. 创建数据库连接连接参数

MyBatis的配置对象Configuration作用及说明

《MyBatis的配置对象Configuration作用及说明》MyBatis的Configuration对象是MyBatis的核心配置对象,它包含了MyBatis运行时所需的几乎所有配置信息,这个对... 目录MyBATis配置对象Configuration作用Configuration 对象的主要作用C

IDEA与JDK、Maven安装配置完整步骤解析

《IDEA与JDK、Maven安装配置完整步骤解析》:本文主要介绍如何安装和配置IDE(IntelliJIDEA),包括IDE的安装步骤、JDK的下载与配置、Maven的安装与配置,以及如何在I... 目录1. IDE安装步骤2.配置操作步骤3. JDK配置下载JDK配置JDK环境变量4. Maven配置下

Springboot的自动配置是什么及注意事项

《Springboot的自动配置是什么及注意事项》SpringBoot的自动配置(Auto-configuration)是指框架根据项目的依赖和应用程序的环境自动配置Spring应用上下文中的Bean... 目录核心概念:自动配置的关键特点:自动配置工作原理:示例:需要注意的点1.默认配置可能不适合所有场景

Nginx配置系统服务&设置环境变量方式

《Nginx配置系统服务&设置环境变量方式》本文介绍了如何将Nginx配置为系统服务并设置环境变量,以便更方便地对Nginx进行操作,通过配置系统服务,可以使用系统命令来启动、停止或重新加载Nginx... 目录1.Nginx操作问题2.配置系统服android务3.设置环境变量总结1.Nginx操作问题

Keepalived+Nginx双机配置小结

《Keepalived+Nginx双机配置小结》本文主要介绍了Keepalived+Nginx双机配置小结,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面... 目录1.1 软硬件要求1.2 部署前服务器配置调优1.3 Nginx+Keepalived部署1.3

Apache伪静态(Rewrite).htaccess文件详解与配置技巧

《Apache伪静态(Rewrite).htaccess文件详解与配置技巧》Apache伪静态(Rewrite).htaccess是一个纯文本文件,它里面存放着Apache服务器配置相关的指令,主要的... 一、.htAccess的基本作用.htaccess是一个纯文本文件,它里面存放着Apache服务器

nginx配置多域名共用服务器80端口

《nginx配置多域名共用服务器80端口》本文主要介绍了配置Nginx.conf文件,使得同一台服务器上的服务程序能够根据域名分发到相应的端口进行处理,从而实现用户通过abc.com或xyz.com直... 多个域名,比如两个域名,这两个域名其实共用一台服务器(意味着域名解析到同一个IP),一个域名为abc

nginx生成自签名SSL证书配置HTTPS的实现

《nginx生成自签名SSL证书配置HTTPS的实现》本文主要介绍在Nginx中生成自签名SSL证书并配置HTTPS,包括安装Nginx、创建证书、配置证书以及测试访问,具有一定的参考价值,感兴趣的可... 目录一、安装nginx二、创建证书三、配置证书并验证四、测试一、安装nginxnginx必须有"-