Flink优化----数据倾斜

devtools/2024/12/22 16:42:14/

目录

判断是否存在数据倾斜

数据倾斜的解决

keyBy 后的聚合操作存在数据倾斜

为什么不能直接用二次聚合来处理

使用 LocalKeyBy 的思想

DataStream API 自定义实现的案例

keyBy 之前发生数据倾斜

keyBy 后的窗口聚合操作存在数据倾斜

实现思路

提交原始案例

提交两阶段聚合的案例

总结

        在大数据流式处理的场景中，Apache Flink 发挥着极为重要的作用。然而，数据倾斜问题就像隐藏在暗处的绊脚石，常常会对 Flink 作业的性能产生严重影响，导致任务执行效率低下、资源利用不均衡等状况。了解如何精准判断数据倾斜是否存在，并掌握有效的解决策略，对于保障 Flink 作业的平稳、高效运行意义重大。无论是开发人员还是运维人员，都需要深入探究这一关键问题，以便更好地应对实际工作中可能遇到的各种挑战，让我们一同走进 Flink 数据倾斜相关内容的学习。

判断是否存在数据倾斜

        相同 Task 的多个 Subtask 中，个别 Subtask 接收到的数据量明显大于其他 Subtask 接收到的数据量，通过 Flink Web UI 可以精确地看到每个 Subtask 处理了多少数据，即可判断出 Flink 任务是否存在数据倾斜。通常，数据倾斜也会引起反压。

        另外，有时 Checkpoint detail 里不同 SubTask 的 State size 也是一个分析数据倾斜的有用指标。

数据倾斜的解决

keyBy 后的聚合操作存在数据倾斜

提交案例：

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=2048mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.SkewDemo1 \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar \
--local-keyby false

查看 webui：

为什么不能直接用二次聚合来处理

        Flink 是实时流处理，如果 keyby 之后的聚合操作存在数据倾斜，且没有开窗口（没攒批）的情况下，简单的认为使用两阶段聚合，是不能解决问题的。因为这个时候 Flink 是来一条处理一条，且向下游发送一条结果，对于原来 keyby 的维度（第二阶段聚合）来讲，数据量并没有减少，且结果重复计算（非 FlinkSQL，未使用回撤流），如下图所示：

使用 LocalKeyBy 的思想

        在 keyBy 上游算子数据发送之前，首先在上游算子的本地对数据进行聚合后，再发送到下游，使下游接收到的数据量大大减少，从而使得 keyBy 之后的聚合操作不再是任务的瓶颈。类似 MapReduce 中 Combiner 的思想，但是这要求聚合操作必须是多条数据或者一批数据才能聚合，单条数据没有办法通过聚合来减少数据量。从 Flink LocalKeyBy 实现原理来讲，必然会存在一个积攒批次的过程，在上游算子中必须攒够一定的数据量，对这些数据聚合后再发送到下游。
实现方式：

  • DataStreamAPI 需要自己写代码实现
  • SQL 可以指定参数，开启 miniBatch 和 LocalGlobal 功能（推荐，后续介绍）

DataStream API 自定义实现的案例

以计算每个 mid 出现的次数为例，keyby 之前，使用 flatMap 实现 LocalKeyby 功能

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.util.Collector;import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;public class LocalKeyByFlatMapFunc extends RichFlatMapFunction<Tuple2<String, Long>, Tuple2<String, Long>> implements CheckpointedFunction {//Checkpoint 时为了保证 Exactly Once，将 buffer 中的数据保存到该 ListState 中private ListState<Tuple2<String, Long>> listState;//本地 buffer，存放 local 端缓存的 mid 的 count 信息private HashMap<String, Long> localBuffer;//缓存的数据量大小，即：缓存多少数据再向下游发送private int batchSize;//计数器，获取当前批次接收的数据量private AtomicInteger currentSize;//构造器，批次大小传参public LocalKeyByFlatMapFunc(int batchSize) {this.batchSize = batchSize;}@Overridepublic void flatMap(Tuple2<String, Long> value, Collector<Tuple2<String, Long>> out) throws Exception {// 1、将新来的数据添加到 buffer 中Long count = localBuffer.getOrDefault(value, 0L);localBuffer.put(value.f0, count + 1);// 2、如果到达设定的批次，则将 buffer 中的数据发送到下游if (currentSize.incrementAndGet() >= batchSize) {// 2.1 遍历 Buffer 中数据，发送到下游for (Map.Entry<String, Long> midAndCount : localBuffer.entrySet()) {out.collect(Tuple2.of(midAndCount.getKey(), midAndCount.getValue()));}// 2.2 Buffer 清空，计数器清零localBuffer.clear();currentSize.set(0);}}@Overridepublic void snapshotState(FunctionSnapshotContext context) throws Exception {// 将 buffer 中的数据保存到状态中，来保证 Exactly OncelistState.clear();for (Map.Entry<String, Long> midAndCount : localBuffer.entrySet()) {listState.add(Tuple2.of(midAndCount.getKey(), midAndCount.getValue()));}}@Overridepublic void initializeState(FunctionInitializationContext context) throws Exception {// 从状态中恢复 buffer 中的数据listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<Tuple2<String, Long>>("localBufferState",Types.TUPLE(Types.STRING, Types.LONG)));localBuffer = new HashMap();if (context.isRestored()) {// 从状态中恢复数据到 buffer 中for (Tuple2<String, Long> midAndCount : listState.get()) {// 如果出现 pv!= 0,说明改变了并行度，ListState 中的数据会被均匀分发到新的 subtask中// 单个 subtask 恢复的状态中可能包含多个相同的 mid 的 count数据// 所以每次先取一下buffer的值，累加再putlong count = localBuffer.getOrDefault(midAndCount.f0, 0L);localBuffer.put(midAndCount.f0, count + midAndCount.f1);}// 从状态恢复时，默认认为 buffer 中数据量达到了 batchSize，需要向下游发currentSize = new AtomicInteger(batchSize);} else {currentSize = new AtomicInteger(0);}}
}

提交 localkeyby 案例：

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=2048mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.SkewDemo1 \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar \
--local-keyby true

查看 webui：

可以看到每个 subtask 处理的数据量基本均衡，另外处理的数据量相比原先少了很多。

keyBy 之前发生数据倾斜

        如果 keyBy 之前就存在数据倾斜，上游算子的某些实例可能处理的数据较多，某些实例可能处理的数据较少，产生该情况可能是因为数据源的数据本身就不均匀，例如由于某些原因 Kafka 的 topic 中某些 partition 的数据量较大，某些 partition 的数据量较少。对于不存在 keyBy 的 Flink 任务也会出现该情况。

        这种情况，需要让 Flink 任务强制进行 shuffle。使用 shuffle、rebalance 或 rescale 算子即可将数据均匀分配，从而解决数据倾斜的问题。

keyBy 后的窗口聚合操作存在数据倾斜

        因为使用了窗口，变成了有界数据（攒批）的处理，窗口默认是触发时才会输出一条结果发往下游，所以可以使用两阶段聚合的方式：

实现思路

  • 第一阶段聚合：key 拼接随机数前缀或后缀，进行 keyby、开窗、聚合
    注意：聚合完不再是 WindowedStream，要获取 WindowEnd 作为窗口标记作为第二阶段分组依据，避免不同窗口的结果聚合到一起）
  • 第二阶段聚合：按照原来的 key 及 windowEnd 作 keyby、聚合

提交原始案例

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=2048mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.SkewDemo2 \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar \
--two-phase false

查看 WebUI：

提交两阶段聚合的案例

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \
-Drest.flamegraph.enabled=true \
-Dyarn.application.queue=test \
-Djobmanager.memory.process.size=1024mb \
-Dtaskmanager.memory.process.size=2048mb \
-Dtaskmanager.numberOfTaskSlots=2 \
-c com.atguigu.flink.tuning.SkewDemo2 \
/opt/module/flink-1.13.1/myjar/flink-tuning-1.0-SNAPSHOT.jar \
--two-phase true \
--random-num 16

查看 WebUI：可以看到第一次打散的窗口聚合，比较均匀

第二次聚合，也比较均匀：

随机数范围，需要自己去测，因为 keyby 的分区器是（两次 hash * 下游并行度 / 最大并行度）
SQL 写法参考：https://zhuanlan.zhihu.com/p/197299746

总结

        本文聚焦 Flink 数据倾斜这一关键要点，系统地介绍了判断与解决相关问题的重要知识。

        在判断数据倾斜方面，明确指出可依据 Flink Web UI 中 Subtask 的数据处理量差异，以及 Checkpoint detail 里 SubTask 的 State size 情况来判断是否存在数据倾斜，为及时发现问题提供了有效途径。

        而在解决数据倾斜问题上，针对 keyBy 后的聚合操作存在倾斜，介绍了通过代码实现（如 DataStreamAPI 自定义代码及 SQL 开启相关功能）来均衡数据量；对于 keyBy 之前发生倾斜，利用 shuffle、rebalance 或 rescale 算子强制数据均匀分配；keyBy 后的窗口聚合操作倾斜时，则采用两阶段聚合的方式，详细说明了各阶段的操作要点及注意事项，并给出相应案例查看效果。

        总之，掌握这些数据倾斜相关的判断方法与解决策略，能帮助使用者更好地优化 Flink 作业，提升整体性能，避免因数据倾斜带来的诸多不良影响，保障数据处理工作顺利开展。


http://www.ppmy.cn/devtools/144420.html

相关文章

【原生js案例】ajax的简易封装实现后端数据交互

【原生js案例】ajax的简易封装实现后端数据交互

ajax是前端与后端数据库进行交互的最基础的工具&#xff0c;第三方的工具库比如jquery,axios都有对ajax进行第二次的封装&#xff0c;fecth是浏览器原生自带的功能&#xff0c;但是它与ajax还是有区别的&#xff0c;总结如下&#xff1a; ajax与fetch对比 实现效果 代码实现 …
阅读更多...
深入解析PCIe地址空间与寄存器机制：从地址映射到TLP生成的完整流程

深入解析PCIe地址空间与寄存器机制：从地址映射到TLP生成的完整流程

往期内容 本文章相关专栏往期内容&#xff0c;PCI/PCIe子系统专栏&#xff1a; 嵌入式系统的内存访问和总线通信机制解析、PCI/PCIe引入 深入解析非桥PCI设备的访问和配置方法 PCI桥设备的访问方法、软件角度讲解PCIe设备的硬件结构 深入解析PCIe设备事务层与配置过程 PCIe的三…
阅读更多...
Sentinel 学习笔记3-责任链与工作流程

Sentinel 学习笔记3-责任链与工作流程

本文属于sentinel学习笔记系列。网上看到吴就业老师的专栏&#xff0c;原文地址如下&#xff1a; https://blog.csdn.net/baidu_28523317/category_10400605.html 上一篇梳理了概念与核心类&#xff1a;Sentinel 学习笔记2- 概念与核心类介绍-CSDN博客 补一个点&#xff1a;…
阅读更多...
【pytorch】多层感知机

【pytorch】多层感知机

将许多全连接层堆叠在一起。每一层都输出到上面的层&#xff0c;直到生成最后的输出。我们可以把前L−1层看作表示&#xff0c;把最后一层看作线性预测器。这种架构通常称为多层感知机通常缩写为MLP。 1 激活函数 激活函数&#xff08;activation function&#xff09;通过计…
阅读更多...
【腾讯云】AI驱动TDSQL-C Serveress 数据库技术实战营-如何是从0到1体验电商可视化分析小助手得统计功能，一句话就能输出目标统计图

【腾讯云】AI驱动TDSQL-C Serveress 数据库技术实战营-如何是从0到1体验电商可视化分析小助手得统计功能，一句话就能输出目标统计图

欢迎来到《小5讲堂》 这是《腾讯云》系列文章&#xff0c;每篇文章将以博主理解的角度展开讲解。 温馨提示&#xff1a;博主能力有限&#xff0c;理解水平有限&#xff0c;若有不对之处望指正&#xff01; 目录 背景效果图流程图创建数据库基本信息数据库配置设置密码控制台开启…
阅读更多...
macOS brew安装

macOS brew安装

brew 可以用命令在mac上安装、卸载、更新各种软件包。它是一个用ruby写的软件&#xff0c;软件是托管在github上的。 1、安装brew&#xff0c;一般选择清华源&#xff0c;如果之前安装过brew&#xff0c;执行下面命令时会自动备份old_homebrew&#xff0c;按照操作执行即可&…
阅读更多...
Datawhale AI冬令营——Chat-悟空设计

Datawhale AI冬令营——Chat-悟空设计

Chat - 悟空项目介绍 一、项目背景 当前大模型市场竞争激烈&#xff0c;通用大模型众多&#xff0c;但针对特定领域、具有特色风格的垂直领域微调模型仍有较大发展空间。以《西游记》这一高人气影视IP为依托进行微调&#xff0c;能在文化娱乐相关细分市场吸引用户关注&#xf…
阅读更多...
Unity 6 Preview（预览版）新增功能

Unity 6 Preview（预览版）新增功能

原文链接&#xff1a;Unity - 手册&#xff1a;Unity 6 预览版中的新增功能 目录 原文链接&#xff1a;Unity - 手册&#xff1a;Unity 6 预览版中的新增功能 编辑器和工作流程 UI 工具包 实体 图形 URP HDRP &#xff08;HDRP&#xff09; 多人游戏 游戏对象的 Netc…
阅读更多...
最新文章

Copyright @ 2022~2023