Flink:Partitioner(分区规则)
Flink之Partitioner(分区规则)
方法 | 注释 |
---|---|
global() | 全部发往1个task |
broadcast() | 广播(前面的文章讲解过,这里不做阐述) |
forward() | 上下游并行度一致时一对一发送,和同一个算子连中算子的OneToOne是一回事 |
shuffle() | 随机分配(只是随机,同Spark的shuffle不同) |
rebalance() | 轮询分配,默认机制就是rebalance() |
recale() | 一般是下游task是上游task的并行度的倍数时,在生成job时,会将下游中的某几个subtask和上游的某个subtask绑成一组,然后在组内上游subtask以轮询的方式将数据发送给下游的subtask. |
partitionCustom | 自定义分区器(这里不做演示,后续会单独写一个自定义分区器的内容) |
keyBy() | 根据数据key的HashCode进行Hash分配 |
-
global
global
在实际业务场景中使用的不是很多,一般都是需要全局数据汇总的时候才会用到.global
就是将上游的数据全部发往下游的第一个subtask
中,也就是说下游设置再多的并行度是没意义的,所以使用global
的时候,下游的task
的并行度都是1
.
这里结合代码看一下:public class FlinkPartitioner { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.setInteger("rest.port", 8081); // 开启本地WebUI,构建流环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf); // 添加数据源,Socket DataStreamSource<String> sourceStream = env.socketTextStream("localhost", 9999); // 转大写,设置并行度为3,且设置数据分区方式为global DataStream<String> upperCaseMapStream = sourceStream.map(s -> s.toUpperCase()).setParallelism(3).global(); // 切分字符串,设置并行度为1 SingleOutputStreamOperator<String> splitFlatMapStream = upperCaseMapStream.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { String[] split = value.split(","); for (String s : split) { out.collect(s); } } }).setParallelism(1); //...... env.execute("Flink Partitioner"); } }
WebUI
界面查看代码中upperCaseMapStream
和splitFlatMapStream之间数据的发送方式
-
forward
forward
其实就是一对一发送数据,和之前讲解Task
的文章中提到的算子之间OneToOne
的模式是一样的,就是可以将forward
理解为同一个task chain
[算子链]中算子之间的数据传输方式,但是使用forward
的前提是上下游的算子并行度是一致的也就是上下游的subtask
数量保持一致,图解如下:代码内容如下:
public class FlinkPartitioner { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.setInteger("rest.port", 8081); // 开启本地WebUI,构建流环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf); // 添加数据源,Socket DataStreamSource<String> sourceStream = env.socketTextStream("localhost", 9999); // 转大写,设置为forward分区方式 DataStream<String> upperCaseMapStream = sourceStream.map(s -> s.toUpperCase()).setParallelism(3).forward(); // 切分字符串 SingleOutputStreamOperator<String> splitFlatMapStream = upperCaseMapStream.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { String[] split = value.split(","); for (String s : split) { out.collect(s); } } }).setParallelism(3).startNewChain(); // 这里加上.startNewChain是为了在WebUI能看到效果,因为upperCaseMapStream和splitFlatMapStream的并行度是一致的,不加startNewChain默认的机制会将两者划分到同一个算子链中,就看不到实际的效果了. // ... env.execute("Flink Partitioner"); } }
WebUI
界面查看upperCaseMapStream
和splitFlatMapStream
的数据发送方式,如下: -
shuffle
通过前面
WebUI
的图片我们可以看到,从Socket
数据源将数据发送到第一个map
的时候使用的是默认的rebalance
方式,也就是轮询发送的方式,而这里说的shuffle
虽然也是一对多的发送方式,但是发送数据时是随机的,举个例子,上游有3
个subtask
,下游有5
个subtask
,数据源有5
条数据,上游中的某一个subtask
向下游发送数据时,是随机发送的,下游的5
个subtask
并不是每个都一定能接受到数据,可能有的接收到1
条,有的接收到2
条,有的接收到3
条数据,这就是shuffle
发送数据的方式.如果说上两个
operator
并行度一致,上游选择了shuffle
发送数据的方式,那么两个operator
会绑定成一个task chain
么?不会,因为shuffle
的数据发送方式就已经导致两个operator
不是OneToOne
的模式了.
代码示例:public class FlinkPartitioner { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.setInteger("rest.port", 8081); // 开启本地WebUI,构建流环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf); // 添加数据源,Socket DataStreamSource<String> sourceStream = env.socketTextStream("localhost", 9999); // 转大写,设置为shuffle分区方式 DataStream<String> upperCaseMapStream = sourceStream.map(s -> s.toUpperCase()).setParallelism(3).shuffle(); // 切分字符串 SingleOutputStreamOperator<String> splitFlatMapStream = upperCaseMapStream.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { String[] split = value.split(","); for (String s : split) { out.collect(s); } } }).setParallelism(7) // ... env.execute("Flink Partitioner"); } }
WebUI
界面查看upperCaseMapStream
和splitFlatMapStream
的数据发送方式,如下: -
Rebalance
rebalance
就是Flink默认的数据分发机制,直白的讲就是给每个小朋友一人一块糖果,直到发完为止,不偏不倚,这个不细说了,没什么可讲的. -
recale
关于
recale
前面说到了是组内的方式进行轮询分发数据,这里就以图解的方式进行讲解,便于理解.Flink任务启动时,如果发现上下游中使用了recale分发数据的方式就会将上下游的subtask进行分组绑定,如上游有2个subtask,下游有四个subtask,就会将上游的一个subtask和下游的两个subtask进行绑定,如下图:
当上下游对应的
subtask
分组后,上下游组内的subtak
就会以组内轮询的方式发送数据,如下图: -
keyBy
keyBy
使用的HASH
分区方式,实际是hashCode()
murmurHash()
的组合方式,这个在源码的KeyGroupRangeAssignment
类中是可以看到的,简单来说根据key
的hash
值模除以下游的最大并行度(return MathUtils.murmurHash(keyHash) % maxParallelism;
).关于
keyBy
的使用应该都很熟悉了,这里直接给大家看演示结果吧,如下图:
这篇好文章是转载于:学新通技术网
- 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
- 本站站名: 学新通技术网
- 本文地址: /boutique/detail/tanhihcebb
-
photoshop保存的图片太大微信发不了怎么办
PHP中文网 06-15 -
photoshop扩展功能面板显示灰色怎么办
PHP中文网 06-14 -
word里面弄一个表格后上面的标题会跑到下面怎么办
PHP中文网 06-20 -
TikTok加速器哪个好免费的TK加速器推荐
TK小达人 10-01 -
《学习通》视频自动暂停处理方法
HelloWorld317 07-05 -
excel图片置于文字下方的方法
PHP中文网 06-27 -
Android 11 保存文件到外部存储,并分享文件
Luke 10-12 -
微信提示登录环境异常是什么意思原因
PHP中文网 04-09 -
微信运动停用后别人还能看到步数吗
PHP中文网 07-22 -
微信人名旁边有个图标有什么用
PHP中文网 03-11