两个字段位置相反去重_Flink零基础教程:并行度和数据重分布
Flink的Transformation转换主要包括四种:单数据流基本转换、基于Key的分组转换、多数据流转换和数据重分布转换。读者可以使用Flink Scala Shell或者Intellij Idea来进行练习:
- Flink Scala Shell:使用交互式编程环境学习和调试Flink
- Flink 01 | 十分钟搭建第一个Flink应用和本地集群
- Flink算子使用方法及实例演示:map、filter和flatMap
- Flink算子使用方法及实例演示:keyBy、reduce和aggregations
- Flink算子使用方法及实例演示:union和connect
并行度
Flink使用并行度来定义某个算子被切分为多少个算子子任务。我们编写的大部分Transformation转换操作能够形成一个逻辑视图,当实际运行时,逻辑视图中的算子会被并行切分为一到多个算子子任务,每个算子子任务处理一部分数据。如下图所示,各个算子并行地在多个子任务上执行,假如算子的并行度为2,那么它有两个实例。
Flink并行执行示意图
并行度可以在一个Flink作业的执行环境层面统一设置,这样将设置该作业所有算子并行度,也可以对某个算子单独设置其并行度。如果不进行任何设置,默认情况下,一个作业所有算子的并行度会依赖于这个作业的执行环境。如果一个作业在本地执行,那么并行度默认是本机CPU核心数。当我们将作业提交到Flink集群时,需要使用提交作业的客户端,并指定一系列参数,其中一个参数就是并行度。
下面的代码展示了如何获取执行环境的默认并行度,如何更改执行环境的并行度。
val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment// 获取当前执行环境的默认并行度val defaultParallelism = senv.getParallelism// 设置所有算子的并行度为4,表示所有算子的并行执行的实例数为4senv.setParallelism(4)
也可以对某个算子设置并行度:
dataStream.map(new MyMapper).setParallelism(defaultParallelism * 2)
数据重分布
默认情况下,数据是自动分配到多个实例上的。有的时候,我们需要手动对数据在多个实例上进行分配,例如,我们知道某个实例上的数据过多,其他实例上的数据稀疏,产生了数据倾斜,这时我们需要将数据均匀分布到各个实例上,以避免部分实例负载过重。数据倾斜问题会导致整个作业的计算时间过长或者内存不足等问题。
下文涉及到的各个数据重分布算子的输入是DataStream,输出也是DataStream。keyBy也有对数据进行分组和数据重分布的功能,但keyBy输出的是KeyedStream。
shuffle
shuffle基于正态分布,将数据随机分配到下游各算子实例上。
dataStream.shuffle()
rebalance与rescale
rebalance使用Round-ribon思想将数据均匀分配到各实例上。Round-ribon是负载均衡领域经常使用的均匀分配的方法,上游的数据会轮询式地分配到下游的所有的实例上。如下图所示,上游的算子会将数据依次发送给下游所有算子实例。
rebalance将数据轮询式地分配到下游实例上
dataStream.rebalance()
rescale与rebalance很像,也是将数据均匀分布到各下游各实例上,但它的传输开销更小,因为rescale并不是将每个数据轮询地发送给下游每个实例,而是就近发送给下游实例。
dataStream.rescale()
上游两个实例下游四个实例时进行rescale
如上图所示,当上游有两个实例时,上游第一个实例将数据发送给下游第一个和第二个实例,上游第二个实例将数据发送给下游第三个和第四个实例,相比rebalance将数据发送给下游每个实例,rescale的传输开销更小。下图则展示了当上游有四个实例,上游前两个实例将数据发送给下游第一个实例,上游后两个实例将数据发送给下游第二个实例。
上游四个实例下游两个实例时进行rescale
broadcast
英文单词"broadcast"翻译过来为广播,在Flink里,数据会被复制并广播发送给下游的所有实例上。
dataStream.broadcast()
global
global会所有数据发送给下游算子的第一个实例上,使用这个算子时要小心,以免造成严重的性能问题。
dataStream.global()
partitionCustom
我们也可以使用partitionCustom来自定义数据重分布逻辑。partitionCustom有两个参数:第一个参数是自定义的Partitioner,我们需要重写里面的partition函数;第二个参数是对数据流哪个字段使用partiton逻辑。partition函数的返回一个整数,表示该元素将被路由到下游第几个实例。
Partitioner[T]中泛型T为指定的字段类型,比如我们要对case class (id: Long, name: String, score: Double)这个数据结构按照id均匀分配到下游各实例,那么泛型T就为id的数据类型Long。同时,泛型T也是partition(key, numPartitions)函数的第一个参数的数据类型。在调用partitionCustom(partitioner, field)时,第一个参数是我们重写的Partitioner,第二个参数表示按照id字段进行处理。
下面的代码按照数据流中的第二个字段进行数据重分布,当该字段中包含数字时,将被路由到下游算子的前半部分,否则被路由到后半部分。如果设置并行度为4,表示所有算子的实例数为4,或者说有4个分区,那么如果字符串包含数字时,该元素将被分配到第0个和第1个实例上,否则被分配到第2个和第3个实例上。
package com.flink.tutorials.api.transformationsimport org.apache.flink.api.common.functions.Partitionerimport org.apache.flink.streaming.api.scala._object PartitionCustomExample { /** * Partitioner[T] 其中泛型T为指定的字段类型 * 重写partiton函数,并根据T字段对数据流中的所有元素进行数据重分配 * */ class MyPartitioner extends Partitioner[String] { val rand = scala.util.Random /** * key 泛型T 即根据哪个字段进行数据重分配,本例中是(Int, String)中的String * numPartitons 为当前有多少个并行实例 * 函数返回值是一个Int 为该元素将被发送给下游第几个实例 * */ override def partition(key: String, numPartitions: Int): Int = { var randomNum = rand.nextInt(numPartitions / 2) // 如果字符串中包含数字,该元素将被路由到前半部分,否则将被路由到后半部分。 if (key.exists(_.isDigit)) { return randomNum } else { return randomNum + numPartitions / 2 } } } def main(args: Array[String]): Unit = { val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // 获取当前执行环境的默认并行度 val defaultParalleism = senv.getParallelism // 设置所有算子的并行度为4,表示所有算子的并行执行的实例数为4 senv.setParallelism(4) val dataStream: DataStream[(Int, String)] = senv.fromElements((1, "123"), (2, "abc"), (3, "256"), (4, "zyx") , (5, "bcd"), (6, "666")) // 对(Int, String)中的第二个字段使用 MyPartitioner 中的重分布逻辑 val partitioned = dataStream.partitionCustom(new MyPartitioner, 1) partitioned.print() senv.execute("partition custom transformation") }}
两个字段位置相反去重_Flink零基础教程:并行度和数据重分布相关推荐
- 零基础学Java大数据难不难
java大数据如今在企业中用到的次数是非常多的,很多人都比较看好java技术,那么零基础学Java大数据难不难?想要学习java技术说难不难,说简单也不是很简单,来看看下面的详细介绍就知道了. 零基础 ...
- 零基础想学大数据?别急!先搞清这一点
在入行大数据开发之前,相信很多同学都存在一些疑虑. 现在入行大数据有前途吗? 学大数据一定要会Java开发吗? 我是零基础,对大数据什么都不懂,能学好吗? .... 大数据时代,给想从事IT行业的人带 ...
- Make 命令零基础教程
Make这个词,英语的意思是"制作".Make命令直接用了这个意思,就是要做出某个文件.比如,要做出文件a.txt,就可以执行下面的命令. $ make a.txt 但是,如果你真 ...
- OpenMV零基础教程
一.资料导航 "工欲善其事,必先利其器".在正式学习OpenMV之前,你必须知道一条或几条OpenMV的学习途径.这里推荐星瞳科技的中文官网教程,这个教程里面包括了OpenMV I ...
- 《网络安全》零基础教程-适合小白科普
<网络安全>零基础教程 目录 目录 <网络安全>零基础教程 第1章 网络安全基础 什么是网络安全 常见的网络安全威胁 网络安全的三个基本要素 网络安全的保障措施 第2章 网络攻 ...
- vue2+vue3小白零基础教程—vue2篇,全网2021最详细教程
vue教程 提示:Vue3系列请参考Vue2+Vue3小白零基础教程-vue3篇文章,本文为vue2篇. 1. Vue核心 1.1 Vue简介 1.1.1 Vue是什么 一套用于构建用户界面的渐进式J ...
- 零基础能学大数据开发吗 可以从哪些方面入手
大数据作为当下呼声特别高的IT技术,想学大数据的朋友已经从一个变成两个,从两个变成三个,但是计数单位,也是从个到百到千到万到亿,接下来还可能更高.零基础能学大数据吗?郑州大数据培训哪家好? 面对这个问 ...
- 适合Java零基础小白学习的Java零基础教程
很多Java零基础小白,在刚刚快入门的时候玩命的学习,玩命的记住Java原理,天天早上五点起床背Java的一些英文词汇,然后遇见一些未知的困难,让自己打到癫狂状态,逐渐迷失自我放弃Java,为了解决这 ...
- NFT潮鞋AR互动零基础教程来啦!
NFT数字藏品,已经成为一种风靡全球的文化潮流.人们不仅能够在元宇宙中拥有数字藏品,还可以利用AR增强现实技术将它们带到现实世界: 今天,我们为大家准备了一份NFT潮鞋定制与AR体验的零基础教程,让你 ...
最新文章
- 将日志中的指定字段对齐显示输出
- 数据链路层中的LLC
- zookeeper系列之通信模型(转)
- matlab 平滑曲线连接_MATLAB数字图像处理-识别广告牌上的文字
- 只需三步即可将 Python 程序转换成 exe 文件
- 计算机配置更新怎么关闭,如何关闭戴尔电脑自动更新系统配置
- 2-1 组合优化问题
- 全国地址json android,全国城市+四级城市地址+邮编+区号+经纬度json版
- 听我给你普及师父、师傅和讲师、教师的区别
- outlook设置京东邮箱
- linux下redis设置密码登录(简单易懂)
- 数数(数学题运算分配律)
- mongodb 分片集群安装 -- 二进制文件安装
- mysql是大端小端_大端和小端 - HackerVirus - 博客园
- 使用KNN和SVM算法实现手写字体识别分类
- fx5u以太网通讯设置_图文简述三菱FX 5U以太网通讯的8大功能,你会用几种?
- 发那科机器人plc电池_FANUC机器人维修保养故障简析
- 从零开始快速搭建SpringBoot+Mybatis+小程序应用--微信小程序的入门和前后端的联调
- 多个Excel文件如何根据条件进行汇总求和呢---多个文件根据条件汇总求和工具
- 左右可以滑动半年的超级日历,支持日历部分收起和自动重定向为北京时间
热门文章
- c语言突然出现图片,c语言能显示图片吗
- c语言学习-使用指针求一个字符串的长度
- php短链接api,PHP通过调用新浪API生成t.cn格式短网址链接的方法详解
- Blazor确认复选框组件
- .NET库和向后兼容的技巧——第1部分
- 数字能排序字符串不能排序_动图解说堆排序原理,让体育生也能看得明白
- 《计算机网络基础》第一套作业,东财在线21春《计算机网络基础》第一套作业【标准答案】...
- java里面的string类型,java中的String类型(不知道理解的好不,请教大神)
- 事务控制 新增后修改_分布式事务科普(初识篇)
- halcon裁剪图像_Halcon学习之七:改变图像的现实方式和大小