FlinkSQL【分组聚合-多维分析-性能调优】应用实例分析

FlinkSQL处理如下实时数据需求：
实时聚合不同 类型/账号/发布时间 的各个指标数据，比如：初始化/初始化后删除/初始化后取消/推送/成功/失败 的指标数据。要求实时产出指标数据，数据源是mysql cdc binlog数据。

代码实例

--SET table.exec.state.ttl=86400s; --24 hour,默认: 0 ms
SET table.exec.state.ttl=2592000s; --30 days,默认: 0 ms
--MiniBatch 聚合
SET table.exec.mini-batch.enabled = true;
SET table.exec.mini-batch.allow-latency = 1s;
SET table.exec.mini-batch.size = 10000;
--Local-Global 聚合
SET table.optimizer.agg-phase-strategy = TWO_PHASE;CREATE TABLE kafka_table (mid bigint,db string,sch string,tab string,opt string,ts bigint,ddl string,err string,src map<string,string>,cur map<string,string>,cus map<string,string>,account_id AS IF(cur['account_id'] IS NOT NULL , cur['account_id'], src ['account_id']),publish_time AS IF(cur['publish_time'] IS NOT NULL , cur['publish_time'], src ['publish_time']),msg_status AS IF(cur['msg_status'] IS NOT NULL , cur['msg_status'], src ['msg_status']),send_type AS IF(cur['send_type'] IS NOT NULL , cur['send_type'], src ['send_type'])--event_time as cast(IF(cur['update_time'] IS NOT NULL , cur['update_time'], src ['update_time']) AS TIMESTAMP(3)), -- TIMESTAMP(3)/TIMESTAMP_LTZ(3)--WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE     --SECOND
) WITH ('connector' = 'kafka','topic' = 't1','properties.bootstrap.servers' = 'xx.xx.xx.xx:9092','properties.group.id' = 'g1','scan.startup.mode' = 'earliest-offset',  --group-offsets/earliest-offset/latest-offset--  'properties.enable.auto.commit',= 'true' -- default:false, 如果为false，则在发生checkpoint时触发offset提交'format' = 'json'
);CREATE TABLE es_sink(send_type      STRING,account_id     STRING,publish_time   STRING,grouping_id       INTEGER,init           INTEGER,init_cancel    INTEGER,push          INTEGER,succ           INTEGER,fail           INTEGER,init_delete    INTEGER,update_time    STRING,PRIMARY KEY (group_id,send_type,account_id,publish_time) NOT ENFORCED
)
with ('connector' = 'elasticsearch-6','index' = 'es_sink','document-type' = 'es_sink','hosts' = 'http://xxx:9200','format' = 'json','filter.null-value'='true','sink.bulk-flush.max-actions' = '1000','sink.bulk-flush.max-size' = '10mb'
);CREATE view  tmp as
selectsend_type,account_id,publish_time,msg_status,case when UPPER(opt) = 'INSERT' and msg_status='0'  then 1 else 0 end AS init,case when UPPER(opt) = 'UPDATE' and send_type='1' and msg_status='4' then 1 else 0 end AS init_cancel,case when UPPER(opt) = 'UPDATE' and msg_status='3' then 1 else 0 end AS push,case when UPPER(opt) = 'UPDATE' and (msg_status='1' or msg_status='5') then 1 else 0 end AS succ,case when UPPER(opt) = 'UPDATE' and (msg_status='2' or msg_status='6') then 1 else 0 end AS fail,case when UPPER(opt) = 'DELETE' and send_type='1' and msg_status='0' then  1 else 0 end AS init_delete,event_time,opt,ts
FROM kafka_table
where (UPPER(opt) = 'INSERT' and msg_status='0' )
or        (UPPER(opt) = 'UPDATE' and msg_status in ('1','2','3','4','5','6'))
or        (UPPER(opt) = 'DELETE' and send_type='1' and msg_status='0');--send_type=1          send_type=0
--初始化->0             初始化->0
--取消->4
--推送->3               推送->3
--成功->1               成功->5
--失败->2               失败->6CREATE view  tmp_groupby as
selectCOALESCE(send_type,'N') AS send_type
,COALESCE(account_id,'N') AS account_id
,COALESCE(publish_time,'N') AS publish_time
,case when send_type is null and account_id is null and publish_time is null then 1when send_type is not null and account_id is null and publish_time is null then 2when send_type is not null and account_id is not null and publish_time is null then 3when send_type is not null and account_id is not null and publish_time is not null then 4end grouping_id
,sum(init) as init
,sum(init_cancel) as init_cancel
,sum(push) as push
,sum(succ) as succ
,sum(fail) as fail
,sum(init_delete) as init_delete
from tmp
--GROUP BY GROUPING SETS ((send_type,account_id,publish_time), (send_type,account_id),(send_type), ())
GROUP BY ROLLUP (send_type,account_id,publish_time); --等同于以上INSERT INTO es_sink
selectsend_type,account_id,publish_time,grouping_id,init,init_cancel,push,succ,fail,init_delete,CAST(LOCALTIMESTAMP AS STRING) as update_time
from tmp_groupby

其他配置

  • flink集群参数
state.backend: rocksdb
state.backend.incremental: true
state.backend.rocksdb.ttl.compaction.filter.enabled: true
state.backend.rocksdb.localdir: /export/io_tmp_dirs/rocksdb
state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints
state.savepoints.dir: hdfs://namenode-host:port/flink-savepoints
rest.flamegraph.enabled: true
pipeline.operator-chaining: false
taskmanager.memory.managed.fraction: 0.7
taskmanager.memory.network.min: 128 mb
taskmanager.memory.network.max: 128 mb
taskmanager.memory.framework.off-heap.size: 32mb
taskmanager.memory.task.off-heap.size: 32mb
taskmanager.memory.jvm-metaspace.size: 256mb
taskmanager.memory.jvm-overhead.fraction: 0.03

  • 检查点配置
    在这里插入图片描述

  • job运行资源
    管理节点(JM) 1 个, 节点规格 1 核 4 GB内存, 磁盘 10Gi
    运行节点(TM）10 个, 节点规格 1 核 4 GB内存, 磁盘 80Gi
    单TM槽位数(Slot): 1
    默认并行度：8

  • es mapping

#POST app_cust_syyy_private_domain_syyy_group_msg/app_cust_syyy_private_domain_syyy_group_msg/_mapping
{"app_cust_syyy_private_domain_syyy_group_msg": {"properties": {"send_type": {"type": "keyword","ignore_above": 256},"account_id": {"type": "keyword"},"publish_time": {"type": "keyword","fields": {"text": {"type": "keyword"},"date": {"type": "date","format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis","ignore_malformed":"true" # 忽略错误的各式}}},"grouping_id": {"type": "integer"},"init": {"type": "integer"},"init_cancel": {"type": "integer"},"query": {"type": "integer"},"succ": {"type": "integer"},"fail": {"type": "integer"},"init_delete": {"type": "integer"},"update_time": {"type": "date","format": "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"}}}
}

性能调优

是否开启【MiniBatch 聚合】和【Local-Global 聚合】对分组聚合场景影响巨大，尤其是在数据量大的场景下。

  • 如果未开启，在分组聚合，数据更新状态时，每条数据都会触发聚合运算，进而更新StateBackend （尤其是对于 RocksDB StateBackend，火焰图上反映就是一直在update rocksdb），造成上游算子背压特别大。此外，生产中非常常见的数据倾斜会使这个问题恶化，并且容易导致 job 发生反压。
    在这里插入图片描述

  • 在开启【MiniBatch 聚合】和【Local-Global 聚合】后，配置如下：

--MiniBatch 聚合
SET table.exec.mini-batch.enabled = true;
SET table.exec.mini-batch.allow-latency = 1s;
SET table.exec.mini-batch.size = 10000;
--Local-Global 聚合
SET table.optimizer.agg-phase-strategy = TWO_PHASE;

开启配置好会在DAG上添加两个环节MiniBatchAssigner和LocalGroupAggregate
在这里插入图片描述

对结果的影响

开启了【MiniBatch 聚合】和【Local-Global 聚合】后，一天处理不完的数据，在10分钟内处理完毕

输出结果

在这里插入图片描述在这里插入图片描述

参考：
Group Aggregation
Streaming Aggregation Performance Tuning

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/628779.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

解决kali beef启动失败解问题

解决kali beef启动失败解问题

只限于出现这个提示的时候使用 卸载 ruby apt remove ruby 卸载 beef apt remove beef-xss 重新安装ruby apt-get install ruby apt-get install ruby-dev libpcap-dev gem install eventmachine 重新安装beef apt-get install beef-xss 弄完以上步骤如果还是不行就重启kali再试…
阅读更多...
LLM之RAG实战（十六）| 使用Llama-2、PgVector和LlamaIndex构建LLM Rag Pipeline

LLM之RAG实战（十六）| 使用Llama-2、PgVector和LlamaIndex构建LLM Rag Pipeline

近年来&#xff0c;大型语言模型&#xff08;LLM&#xff09;取得了显著的进步&#xff0c;然而大模型缺点之一是幻觉问题&#xff0c;即“一本正经的胡说八道”。其中RAG&#xff08;Retrieval Augmented Generation&#xff0c;检索增强生成&#xff09;是解决幻觉比较有效的…
阅读更多...
opencv_角点检测

opencv_角点检测

文章内容 一个opencv检测角点的程序 运行效果 #include <opencv2/opencv.hpp> #include <opencv2/highgui/highgui.hpp> #include <opencv2/imgproc/imgproc.hpp> #include <iostream>using namespace cv; using namespace std;void detectCorners(M…
阅读更多...
数据分析-Pandas如何整合多张数据表

数据分析-Pandas如何整合多张数据表

数据分析-Pandas如何整合多张数据表 数据表&#xff0c;时间序列数据在数据分析建模中很常见&#xff0c;例如天气预报&#xff0c;空气状态监测&#xff0c;股票交易等金融场景。数据分析过程中重新调整&#xff0c;重塑数据表是很重要的技巧&#xff0c;此处选择Titanic数据…
阅读更多...
多测师肖sir___ui自动化测试po框架（升级）

多测师肖sir___ui自动化测试po框架（升级）

ui自动化测试po框架&#xff08;升级&#xff09; po框架 一、ui自动化po框架介绍 &#xff08;1&#xff09;PO是Page Object的缩写&#xff08;pom模型&#xff09; &#xff08;2&#xff09;业务流程与页面元素操作分离的模式&#xff0c;可以简单理解为每个页面下面都有一…
阅读更多...
【linux】visudo

【linux】visudo

碎碎念 visudo命令是用来修改一个叫做 /etc/sudoers 的文件的&#xff0c;用来设置哪些 用户 和 组 可以使用sudo命令。并且使用visudo而不是使用 vi /etc/sudoers 的原因在于&#xff1a;visudo自带了检查功能&#xff0c;可以判断是否存在语法问题&#xff0c;所以更加安全 …
阅读更多...
7.评价预测模型——C指数，NRI，IDI计算

7.评价预测模型——C指数，NRI，IDI计算

目录 基本知识 1. C指数 2. NRI、IDI 二分类资料 1. C指数 C指数计算 比较两个模型C指数 2. NRI 3. IDI 生存资料 1. rms包拟合的生存曲线 C指数 比较两个模型的C指数 2. survival包拟合的生存曲线 C指数 NRI计算 IDI 基本知识 1. C指数 C指数&#xff1a; …
阅读更多...
stm32 - 基础架构

stm32 - 基础架构

stm32 - 基础架构 基础架构外设概念系统结构引脚定义晶振工程 基础架构 外设概念 NVIC &#xff08;内核外设&#xff09; SysTick &#xff08;内核外设&#xff09; 其他是片上外设 系统结构 内核引出三条总线 ICode 指令总线&#xff1a; 连接Flash闪存&#xff08;编写的…
阅读更多...
C# wpf 获取控件刷新的时机

C# wpf 获取控件刷新的时机

文章目录 前言一、为何要获取刷新时机&#xff1f;例子一、隐藏控件后截屏例子二、修改控件大小后做计算 二、如何实现&#xff1f;1.使用动画2.使用TaskCompletionSource 三、完整代码四、使用示例1、隐藏工具条截屏2、修改宽高后获取ActualWidth、ActualHeight 总结 前言 做…
阅读更多...
计算机网络（超详解！） 第二节 数据链路层（上）

计算机网络（超详解！） 第二节 数据链路层（上）

1.数据链路层使用的信道 数据链路层使用的信道主要有以下两种类型&#xff1a; 1.点对点信道&#xff1a;这种信道使用一对一的点对点通信方式。 2.广播信道&#xff1a;这种信道使用一对多的广播通信方式&#xff0c;因此过程比较复杂。广播信道上连接的主机很多&#xff0…
阅读更多...
力扣刷MySQL-第二弹（详细解析）

力扣刷MySQL-第二弹（详细解析）

&#x1f389;欢迎您来到我的MySQL基础复习专栏 ☆* o(≧▽≦)o *☆哈喽~我是小小恶斯法克&#x1f379; ✨博客主页&#xff1a;小小恶斯法克的博客 &#x1f388;该系列文章专栏&#xff1a;力扣刷题讲解-MySQL &#x1f379;文章作者技术和水平很有限&#xff0c;如果文中出…
阅读更多...
华为设备VRRP配置

华为设备VRRP配置

核心代码&#xff1a; 需要对所有虚拟路由器设置&#xff08;要进入到对应的端口&#xff09; vrrp vrid 38 virtual-ip 192.168.10.254 vrrp vrid 38 priority 120 vrrp vrid 38 track int g0/0/1 reduced 30①mac由vrid生成 ②指定虚拟ip ③虚拟ip作为内部主机的网关&#x…
阅读更多...
如何在云端加速缓存构建

如何在云端加速缓存构建

缓存是指将某类数据存储起来以便以后重复使用的过程&#xff0c;它的运用在开发场景中非常普遍。类似于你习惯把最常用的调料放在厨房台面上&#xff0c;而不是橱柜里&#xff0c;这样你在准备大餐时就可以轻松取用。 但对于一个更为技术性、更精确的用例&#xff0c;比如像谷…
阅读更多...
云服务器基于Centos创建个人云盘实践经验分享

云服务器基于Centos创建个人云盘实践经验分享

文章目录 安装运行Cloudreve安装ossfscentos更换yum源 配置ossfs挂载oss存储配置开机启动 配置cloudreve推荐阅读 安装运行Cloudreve 执行如下命令&#xff0c;下载cloudreve安装包。 wget https://labfileapp.oss-cn-hangzhou.aliyuncs.com/cloudreve_3.3.1_linux_amd64.tar…
阅读更多...
C#/WPF 设置和启动Windows屏保程序

C#/WPF 设置和启动Windows屏保程序

前言 我们平时电脑启动的屏保程序其本质也是应用程序&#xff0c;只是后缀名为.scr。所以我们只需要把应用程序后缀改为.scr&#xff0c;然后右键选择安装即可启动我们自己的屏保程序。 屏保注册表参数 设置电脑屏保参数&#xff0c;在个性化设置>锁屏界面>屏幕保护程序设…
阅读更多...
Qt/QML编程之路：slider（34）

Qt/QML编程之路：slider（34）

滑条slider&#xff0c;有时也成为进度条progressbar&#xff0c;在GUI界面中也是经常用到的。 import QtQuick 2.9 import QtQuick.Controls 2.0 import QtQuick.Layouts 1.2ApplicationWindow {id:rootvisible: truewidth: 1920height: 720//title: qsTr("Hello World&q…
阅读更多...
rabbitmq-java基础详解

rabbitmq-java基础详解

一、rabbitmq是什么&#xff1f; 1、MQ定义 MQ&#xff08;Message Queue&#xff09;消息队列 主要解决&#xff1a;异步处理、应用解耦、流量削峰等问题&#xff0c;是分布式系统的重要组件&#xff0c;从而实现高性能&#xff0c;高可用&#xff0c;可伸缩和最终一致性的架…
阅读更多...
如何从命令行运行testng.xml？

如何从命令行运行testng.xml？

目录 创建一个新的java项目并从命令行运行testng.xml 使用命令行运行XML文件 从命令行运行现有maven项目的XML文件 在这篇文章中&#xff0c;我们将使用命令行运行testng.xml。有多种场景需要使用命令行工具运行testng.xml。也许您已经创建了一个maven项目&#xff0c;现在想…
阅读更多...
运筹说 第97期|非线性规划-一维搜索

运筹说 第97期|非线性规划-一维搜索

第二节 一维搜索 通过上期学习&#xff0c;大家已经了解了非线性规划的基本内容&#xff0c;那么如何求解一个非线性规划问题呢&#xff1f;本期小编就带大家来学习用于求解单变量无约束极值问题的方法——一维搜索&#xff0c;该方法也是后面求解更复杂问题的基础。 一、引入…
阅读更多...
FinalShell连接虚拟机2024/1/16

FinalShell连接虚拟机2024/1/16

目录 1.右键虚拟机桌面空白处&#xff0c;选择打开终端&#xff0c;在终端中输入ifconfig命令&#xff0c;查看Linux系统的IP地址&#xff1a;复制。 2.打开FinalShell,点击&#xff08;1&#xff09;号文件夹打开连接管理器&#xff0c;点击&#xff08;2&#xff09;号选择…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023