FlinkSQL 开发经验分享

FlinkSQL开发经验分享

作者:汤包

最近做了几个实时数据开发需求,也不可避免地在使用 Flink 的过程中遇到了一些问题,比如数据倾斜导致的反压、interval join、开窗导致的水位线失效等问题,通过思考并解决这些问题,加深了我对 Flink 原理与机制的理解,因此将这些开发经验分享出来,希望可以帮助到有需要的同学。

下文会介绍 3 个 case 案例,每个 case 都会划分为背景、原因分析和解决方法三部分来进行介绍。

一、Case1: 数据倾斜

数据倾斜无论是在离线还是实时中都会遇到,其定义是:在并行进行数据处理的时候,按照某些 key 划分的数据显著多余其他部分,分布不均匀,导致大量数据集中分布到一台或者某几台计算节点上,使得该部分的处理速度远低于平均计算速度,成为整个数据集处理的瓶颈,从而影响整体计算性能。造成数据倾斜的原因有很多种,如 group by 时的 key 分布不均匀,空值过多、count distinct 等,本文将只介绍 group by + count distinct 这种情况。

1.1 背景

对实时曝光流,实时统计近 24 小时创意的曝光 UV 和 PV。且每分钟更新一次数据。通用的方法就是使用 hop 滑动窗口来进行统计,代码如下:

select    Hselect    HOP_START(        ts        ,interval '1' minute        ,interval '24' hour    ) as window_start    ,HOP_END(        ts        ,interval '1' minute        ,interval '24' hour    ) as window_end    ,creative_id    ,count(distinct uid) as exp_uv  -- 计算曝光UV    ,count(uid) as exp_pv   --计算曝光PVfrom dwd_expos_detailgroup by    hop(        ts        ,interval '1' minute        ,interval '24' hour    )  -- 滑动窗口开窗,窗口范围:近24小时,滑动间隔:每1分钟    ,creative_idOP_START(        ts        ,interval '1' minute        ,interval '24' hour    ) as window_start    ,HOP_END(        ts        ,interval '1' minute        ,interval '24' hour    ) as window_end    ,creative_id    ,count(distinct uid) as exp_uv  -- 计算曝光UV    ,count(uid) as exp_pv   --计算曝光PVfrom dwd_expos_detailgroup by    hop(        ts        ,interval '1' minute        ,interval '24' hour    )  -- 滑动窗口开窗,窗口范围:近24小时,滑动间隔:每1分钟    ,creative_id

复制代码

1.2 问题及原因

问题发现

在上述 flink 程序运行的时候,该窗口聚合算子 GlobalWindowAggregate 出现长时间 busy 的情况,导致上游的算子出现反压,整个 flink 任务长时间延迟。

原因分析

一般面对反压的现象,首先要定位到出现拥堵的算子,在该 case 中,使用窗口聚合计算每个创意 id 对应的 UV 和 PV 时,出现了计算繁忙拥堵的情况。

针对这种情况,最常想到的就是以下两点原因:

  • 数据量较大,但是设置的并发度过小(此任务中该算子的并发度设置为 3)

  • 单个 slot 的 CPU 和内存等计算资源不足

点击拥堵算子,并查看 BackPressure,可以看到虽然并发度设置为 3,但是出现拥堵的只有 subtask0 这一个并发子任务,因此基本上可以排出上述两种猜想,如果还是不放心,可以设置增加并行度至 6,同时提高该算子上的 slot 的内存和 CPU,结果如下:

可以看到依然只有 subtask0 处于计算拥堵的状态,现在可以完全确认是由于 group by 时的 key 上的数据分布不均匀导致的数据倾斜问题。

解决方法

  • 开启 PartialFinal 解决 count distinct 中的热点问题

  • 实现:flink 中提供了针对 count distinct 的自动打散和两阶段聚合,即 PartialFinal 优化。实现方法:在作业运维中增加如下参数设置:

table.optimizer.distinct-agg.split.enabled: true

复制代码

  • 限制:这个参数适用于普通的 GroupAggregate 算子,对于 WindowAggregate 算子目前只适用于新的 Window TVF(窗口表值函数),老的一套 Tumble/Hop/Cumulate window 是不支持的。

由于我们的代码中并没有使用到窗口表值函数,而是直接在 group 中使用了 hop 窗口,因此该方法不适用。

人工对不均匀的 key 进行打散并实现两阶段聚合

  • 思路:增加按 Distinct Key 取模的打散层

  • 实现:

    第一阶段:对 distinct 的字段 uid 取 hash 值,并除以 1024 取模作为 group by 的 key。此时的 group by 分组由于引入了 user_id,因此分组变得均匀。

select        HOP_START(            ts            ,interval '1' minute            ,interval '24' hour        ) as window_start        ,HOP_END(            ts            ,interval '1' minute            ,interval '24' hour        ) as window_end        ,creative_id        ,count(distinct uid) as exp_uv        ,count(uid) as exp_pv    from dwd_expos_detail    group by        hop(            ts            ,interval '1' minute            ,interval '24' hour        )        ,creative_id        ,MOD(HASH_CODE(uid), 1024)

复制代码

  • 第二阶段:对上述结果,再根据 creative_id 字段进行分组,并将 UV 和 PV 的值求和

select    window_start    ,window_end    ,creative_id    ,sum(exp_uv) as exp_uv    ,sum(exp_pv) as exp_pvfrom (    select        HOP_START(            ts            ,interval '1' minute            ,interval '24' hour        ) as window_start        ,HOP_END(            ts            ,interval '1' minute            ,interval '24' hour        ) as window_end        ,creative_id        ,count(distinct uid) as exp_uv        ,count(uid) as exp_pv    from dwd_expos_detail    group by        hop(            ts            ,interval '1' minute            ,interval '24' hour        )        ,creative_id        ,MOD(HASH_CODE(uid), 1024))group by    window_start    ,window_end    ,creative_id;

复制代码

  • 效果:在拓扑图中可以看到原窗口聚合算子被分为两个独立的聚合算子,同时每个 subtask 的繁忙程度也都接近,不再出现不均匀的情况。

二、Case2: 水位线失效

2.1 背景

需要先对两条实时流进行双流 join,然后再对 join 后的结果使用 hop 滑动窗口,计算每个创意的汇总指标。

2.2 问题及原因

问题发现

开窗后长时间无数据产生。

原因分析

水位线对于窗口函数的实现起到了决定性的作用,它决定了窗口的触发时机,Window 聚合目前支持 Event Time 和 Processing Time 两种时间属性定义窗口。最常用的就是在源表的 event_time 字段上定义水位线,系统会根据数据的 Event Time 生成的 Watermark 来进行关窗。

只有当 Watermark 大于关窗时间,才会触发窗口的结束,窗口结束才会输出结果。如果一直没有触发窗口结束的数据流入 Flink,则该窗口就无法输出数据。

  • 限制:数据经过 GroupBy、双流 JOIN 或 OVER 窗口节点后,会导致 Watermark 属性丢失,无法再使用 Event Time 进行开窗。

由于我们在代码中首先使用了 interval join 来处理点击流和交易流,然后在对生成的数据进行开窗,导致水位线丢失,窗口函数无法被触发。

2.3 解决方法

思路 1: 既然双流 join 之后的时间字段丢失了水位线属性,可以考虑再给 join 之后的结果再加上一个 processing time 的时间字段,然后使用该字段进行开窗。

  • 缺点:该字段无法真正体现数据的时间属性,只是机器处理该条数据的时间戳,因此会导致窗口聚合时的结果不准确,不推荐使用。

思路 2: 新建 tt 流

  • 要开窗就必须有水位线,而水位线往往会在上述提及的聚合或者双流 join 加工中丢失,因此考虑新建一个 flink 任务专门用来进行双流 join,过滤出符合条件的用户交易明细流,并写入到 tt,然后再消费该 tt,并对 tt 流中的 event_time 字段定义 watermark 水位线,并直接将数据用于 hop 滑动窗口。

  • 实现:

    步骤 1:新建 flink 任务,通过 interval join 筛选出近六个小时内有过点击记录的用户交易明细,并 sink 到 tt

insert into sink_dwd_pop_pay_detail_riselect    p1.uid    ,p1.order_id    ,p1.order_amount    ,p1.ts    ,p2.creative_idfrom (    select        uid        ,order_amount         ,order_id        ,ts    from dwd_trade_detail) p1    join dwd_clk_uv_detail p2        on p2.ts between p1.ts - interval '6' hour and p1.ts        and p1.uid = p2.uid;

复制代码

  • 步骤 2: 消费该加工后的交易流,并直接进行滑动窗口聚合

select    HOP_START(        ts        ,INTERVAL '1' minute        ,INTERVAL '24' hour    ) as window_start    ,HOP_END(        ts        ,INTERVAL '1' minute        ,INTERVAL '24' hour    ) as window_end    ,creative_id    ,sum(order_amount) as total_gmv    ,count(distinct uid) as cnt_order_uv    ,round(        sum(order_amount) / count(distinct uid) / 1.0        ,2    ) as gmv_per_uvfrom source_dwd_pop_pay_detail_riGROUP BY    HOP(        ts        ,INTERVAL '1' minute        ,INTERVAL '24' hour    )    ,creative_id;

复制代码

三、Case3: group by 失效

3.1 背景

目的:对于实时流,需要给素材打上是否通过的标签。

打标逻辑:如果素材 id 同时出现在 lastValidPlanInfo 和 validPlanInfo 的两个数组字段中,则认为该素材通过(is_filtered=0),如果素材 id 只出现在 lastValidPlanInfo 数组字段中,则认为该素材未通过(is_filtered= 1)。

sink 表类型:odps/sls,不支持回撤和主键更新机制。

上述逻辑的实现 sql 如下:

SELECT    `user_id`    ,trace_id    ,`timestamp`    ,material_id     ,min(is_filtered)) as is_filtered   -- 最后group by聚合,每个素材得到唯一的标签    FROM ( SELECT     `user_id`     ,trace_id     ,`timestamp`     ,material_id     ,1 as is_filtered   -- lastValidPlanInfo字段中出现的素材都打上1的被过滤标签 FROM dwd_log_parsing     ,lateral table(string_split(lastValidPlanInfo, ';')) as t1(material_id) WHERE lastValidPlanInfo IS NOT NULL UNION ALL SELECT     `user_id`     ,trace_id     ,`timestamp`     ,material_id     ,0 as is_filtered     -- validPlanInfo字段中出现的素材都打上0的被过滤标签 FROM dwd_log_parsing   ,lateral table(string_split(validPlanInfo, ';')) as t2(material_id)      WHERE validPlanInfo IS NOT NULL    )    GROUP BY        `user_id`        ,trace_id        ,`timestamp`        ,material_id

复制代码

3.2 问题及原因

问题发现

原始数据样例:根据下图可以发现 1905 和 1906 两个素材 id 出现在 lastValidPlanInfo 中,只有 1906 这个 id 出现在 validPlanInfo 字段中,说明 1905 被过滤掉了,1906 通过了。

期望的计算结果应该是:

但是最终写入到 odps 的结果如下图,可以发现 material_id 为 1906 出现了两条结果,且不一致,所以我们不禁产生了一个疑问:是 fink 中的 group by 失效了吗?

原因分析

由于 odps sink 表不支持回撤和 upsert 主键更新机制,因此对于每一条源表的流数据,只要进入到 operator 算子并产生结果,就会直接将该条结果写入到 odps。

union all 和 lateral table 的使用都会把一条流数据拆分为多条流数据。上述代码中首先使用到了 lateral table 将 lastValidPlanInfo 和 validPlanInfo 数组字段中的 material_id 数字拆分为多条 material_id,然后再使用 union all+group by 实现过滤打标功能,这些操作早已经将原 tt 流中的一条流数据拆分成了多条。

综合上述两点,

  • 针对 1906 的素材 id,由于 lateral table 的使用,使得其和 1905 成为了两条独立的流数据;

  • 由于 union all 的使用,又将其拆分为 is_filtered =1 的一条流数据(union all 的前半部分),和 is_filtered=0 的一条流数据(union all 的后半部分);

  • 由于 flink 一次只能处理一条流数据,因此如果先处理了素材 1906 的 is_filtered=1 的流数据,经过 group by 和 min(is_filtered)操作,将 is_filtered= 1 的结果先写入到 odps,然后再处理 is_filtered=1 的流数据,经过 group by 和 min(is_filtered)操作,状态更新 is_filtered 的最小值变更为 0,又将该条结果写入到 odps。

  • 由于 odps 不支持回撤和主键更新,因此会存在两条素材 1906 的数据,且结果不一致。

3.3 解决方法

  • 思路:既然 lateral table 和 union all 的使用,会把一条流数据变为多条,并引发了后续的多次写入的问题。因此我们考虑让这些衍生出的多条流数据可以一次性进入到 group by 中参与聚合计算,最终只输出 1 条结果。

  • 实现:mini-batch 微批处理

table.exec.mini-batch.enabled: truetable.exec.mini-batch.allow-latency: 1s

复制代码

  • 概念:mini-batch 是缓存一定的数据后再触发处理,以减少对 State 的访问,从而提升吞吐并减少数据的输出量。微批处理通过增加延迟换取高吞吐,如果您有超低延迟的要求,不建议开启微批处理。通常对于聚合场景,微批处理可以显著地提升系统性能,建议开启。

  • 效果:上述问题得到解决,odps 表只输出每个用户的每次请求的每个素材 id 只有 1 条数据输出。

四、总结

FlinkSQL 的开发是最方便高效的实时数据需求的实现途径,但是它和离线的 ODPS SQL 开发在底层的机制和原理上还是有很大的区别,根本的区别就在于流和批的处理。如果按照我们已经习惯的离线思维来写 FlinkSQL,就可能会出现一些“离奇”的结果,但是遇到问题并不可怕,要始终相信根本不存在任何“离奇”,所有的问题都是可以追溯到原因的,而在这个探索的过程中,也可以学习到许多知识,所以让我们遇到更多的问题,积累更多的经验,熟练地应用 Flink。

参考链接

[01] 窗口

https://help.aliyun.com/zh/flink/developer-reference/overview-4?spm=a2c4g.11186623.0.i33

[02] 高性能优化:

https://help.aliyun.com/zh/flink/user-guide/optimize-flink-sql

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

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

相关文章

华为云简介

前言 华为云是华为的云服务品牌,将华为30多年在ICT领域的技术积累和产品解决方案开放给客户,致力于提供稳定可靠、安全可信、可持续创新的云服务,赋能应用、使能数据、做智能世界的“黑土地”,推进实现“用得起、用得好、用得放心…

鸿蒙应用笔记

安装就跳过了,一直点点就可以了 配置跳过,就自动下了点东西。 鸿蒙那个下载要12g个内存,大的有点吓人。 里面跟idea没区别 模拟器或者真机运行 真机要鸿蒙4.0,就可以实机调试 直接在手机里面跑,这个牛逼&#xf…

深度学习与CV入门

文章目录 前言历史 前言 历史 tensorflow可以安装Tensorboard第三方库用于展示效果 TensorFlow工作流程:p6-4:20 使用tf.data加载数据。使用tf.data实例化读取训练数据和测试数据模型的建立与调试:使用动态图模式Eager Execution和著名的神经网络高层API框架Ker…

[笔记] 卷积 - 02 滤波器在时域的等效形式

1.讨论 这里主要对时域和频域的卷积运算的特征做了讨论,特别是狄拉克函数的物理意义。 关于狄拉克函数,参考这个帖子:https://zhuanlan.zhihu.com/p/345809392 1.狄拉克函数提到的好函数的基本特征是能够快速衰减,对吧&#xf…

【雷丰阳-谷粒商城 】【分布式高级篇-微服务架构篇】【17】认证服务01—短信/邮件/异常/MD5

持续学习&持续更新中… 守破离 【雷丰阳-谷粒商城 】【分布式高级篇-微服务架构篇】【17】认证服务01 环境搭建验证码倒计时短信服务邮件服务验证码短信形式:邮件形式: 异常机制MD5参考 环境搭建 C:\Windows\System32\drivers\etc\hosts 192.168.…

HackTheBox----Editorial

Editorial 测试过程 1 信息收集 NMAP端口扫描 nmap -sC -sV 10.10.11.20服务器开启了 22、80 端口 80 端口测试 服务器只开启了 22 和 80 端口,先从 80 端口开始进行测试 echo "10.10.11.20 editorial.htb" | sudo tee -a /etc/hostspublish with us…

Scrapy框架的基本使用教程

1、创建scrapy项目 首先在自己的跟目录文件下执行命令: PS D:\BCprogram\python_pro\bigdata> scrapy startproject theridion_grallatorscrapy startproject 项目名 具体执行操作如下:1、创建项目目录:Scrapy会在当前工作目录下创建一…

001,函数指针是一种特殊的指针,它指向的是一个函数地址,可以存储函数并作为参数传递,也可以用于动态绑定和回调函数

函数指针是一种特殊的指针 001,函数指针是一种特殊的指针,它指向的是一个函数地址,可以存储函数并作为参数传递,也可以用于动态绑定和回调函数 文章目录 函数指针是一种特殊的指针前言总结 前言 这是ai回答的标准答案 下面我们…

bash条件判断基础adsawq1`1nn

判断的作用 判断后续操作的提前条件是否满足如果满足执行一种命令不满足则执行另一种指令 条件测试类型: 整型测试字符测试文字测试 整数测试:比较两个整数谁大谁小,是否相等; 二元测试: num1 操作符 num2 -eq: 等于…

Alt与Tab切换窗口时将Edge多个标签页作为一个整体参与切换的方法

本文介绍在Windows电脑中,使用Alt与Tab切换窗口时,将Edge浏览器作为一个整体参与切换,而不是其中若干个页面参与切换的方法。 最近,需要将主要使用的浏览器由原本的Chrome换为Edge;但是,在更换后发现&#…

基于RK3588的8路摄像头实时全景拼接

基于RK3588的8路摄像头实时全景拼接 输入:2路csi转8路mpi的ahd摄像头,分辨率1920 * 1080 8路拼接结果: 6路拼接结果: UI界面: UI节目设计原理

Lua、AB包热更新总结

1.AB包热更新 (1)AB包是一种特定的压缩文件,可以放模型贴图音效等等 (2)Resources目录下打包时只读 无法修改;而AB包存储的位置是自定义的,能够动态更新,同时可以决定资源包初始的大…

huggingface笔记:gpt2

0 使用的tips GPT-2是一个具有绝对位置嵌入的模型,因此通常建议在输入的右侧而不是左侧填充GPT-2是通过因果语言建模(CLM)目标进行训练的,因此在预测序列中的下一个标记方面非常强大 利用这一特性,GPT-2可以生成语法连…

人工智能时代打工人摸鱼秘籍(1)- 为啥说大模型像人?

人工智能以势不可挡的方式席卷全球。 所有公司,都在削尖脑袋想,如何在在产品、营销、运营、服务和管理上加持大人工智能的能力。 公司在卷生卷死的时候,有一批人已经偷偷在用大模型提(摸)效(鱼)…

昇思25天学习打卡营第1天|初识MindSpore

# 打卡 day1 目录 # 打卡 day1 初识MindSpore 昇思 MindSpore 是什么? 昇思 MindSpore 优势|特点 昇思 MindSpore 不足 官方生态学习地址 初识MindSpore 昇思 MindSpore 是什么? 昇思MindSpore 是全场景深度学习架构,为开发者提供了全…

KDP数据分析实战:从0到1完成数据实时采集处理到可视化

智领云自主研发的开源轻量级Kubernetes数据平台,即Kubernetes Data Platform (简称KDP),能够为用户提供在Kubernetes上的一站式云原生数据集成与开发平台。在最新的v1.1.0版本中,用户可借助 KDP 平台上开箱即用的 Airflow、AirByte、Flink、K…

LLMs之gptpdf:gptpdf的简介、安装和使用方法、案例应用之详细攻略

LLMs之gptpdf:gptpdf的简介、安装和使用方法、案例应用之详细攻略 目录 gptpdf的简介 1、处理流程 第一步,使用 PyMuPDF 库,对 PDF 进行解析出所有非文本区域,并做好标记,比如: 第二步,使用视觉大模型&…

RUST 编程语言 绘制随机颜色图片 画圆形 画矩形 画直线

什么是Rust Rust是一种系统编程语言,旨在提供高性能和安全性。它是由Mozilla和其开发社区创建的开源语言,设计目标是在C的应用场景中提供一种现代、可靠和高效的选择。Rust的目标是成为一种通用编程语言,能够处理各种计算任务,包…

基于若依的文件上传、下载

基于若依实现文件上传、下载 文章目录 基于若依实现文件上传、下载1、前端实现-文件上传1.1 通用上传分析1.2 修改实现上传接口 2、后端实现-文件上传3、后端实现-文件下载4、前端实现-文件下载 官网其实也写了,但是我是自己改造封装了一下,再次迈向全栈…

3.js - 模板渲染 - 简单

3.js 真tm枯燥啊,狗都不学 效果图 源码 // ts-nocheck// 引入three.js import * as THREE from three// 导入轨道控制器 import { OrbitControls } from three/examples/jsm/controls/OrbitControls// 导入lil.gui import { GUI } from three/examples/jsm/libs/li…