00:00
了解了shuffle和reb这两种最基础的分区方式,那接下来呢,我们再来介绍一种比较特殊的用法,哎,那就是所谓的重缩放分区。这种分区方式呢,简单来说跟前面我们介绍的reb轮询是非常相似的,它的底层其实也是Robin,也是做轮询,它呢就是基于data three调一个方法就可以。那它跟reb的区别在哪里呢?诶,我们可以看到啊,前面我们讲到rebance的时候,它其实是把上游每一个并行子任务里边的所有数据全部平均分配到下游的并行子任务里面去了,比方说我们看啊,上游有两个并行子任务,那第一个并行子任务里面的数据ABCDEF。他会第一个A放到下游的第一个里边来,然后B分配到第二个任务里边来,C分配到第三个,那接下来呢,B又分配到第一个,所以我们看到它是针对下游的三个并行子任务。
01:03
全部做了一个轮巡发牌。而现在的呢啊,他不是对所有人轮询发牌了,他是做了一个分组。哎,我们分成了小团体,我只在当前的组内做一个轮巡发牌,所以我们看到啊,上游任务有两个冰行子任务,下游任务有四个并行子任务,那这个时候呢,从中间分开,一分为二,那就每一组内是一个一对二的关系,所以当前上游的第一个并行子任务ABCDEF数据分发到哪里去呢?只分发到下游的前两个并行子任务里面去轮询发放就可以了。啊,同样道理,那上游的第二个并行子任务,它的数据123456,就对应的轮询发放到下游的第三第四个这两个并行子任务里面去。所以我们可以认为它所谓的这个重缩放就是做了分组的一个rebance,一个轮群啊啊,那这里我们会发现啊,这种方式在什么情况下最为有优势,最为有效率呢?很显然就是分组比较均匀的时候,当然就最为有效率了,最好的情况那就是。
02:20
上游任务和下游任务它是整数倍的关系啊,就是比方说我们这里的二对四或者三对六啊,那当然我们这里是两倍关系啊,如果是三倍四倍关系也是可以的,有了这样的整数倍对应关系之后,那我们就知道了,对应的一个上游并行子任务,就可以直接把数据分发到自己当前小组内的几个下游任务就可以了,那这样就直接小范围内轮寻起来了。如果对应到flink的运行式架构里边的话,我们也会想到啊,那当前我们一个flink集群里边,它不是分布式集群吗?那可能会有多个task manager,而每个task manager上呢,又有多个task slots啊,所以这个时候我们就会发现啊,如果你要跨不同的task manager之间要去传递数据的话,那肯定这个代价就比较高嘛啊这里涉及到了网络传输,如果说你只是在当前的一个task manager里边去进行slos之间的数据传递,相对来讲代价就会低一点啊,所以我们看到,那如果是这种情况的话,那我们就干脆按照task manager来做一个分组,然后每个分组内的task slot里边自己去做轮询,这不是效率会更高吗?诶,这是我们能够考虑到的,对于我们实际应用的一种优化,那另外还有一点就是我们前面说的,对于re balance这种重分区方式来讲,它其实是。
03:52
上游下游所有并行子任务之间的一个一一对应的关系啊,那是一个笛卡尔基M乘N,它们之间的数据传输通道要建立M乘N个这样的信道,而现在的rib scale呢,它就相当于我们把这个数量级直接减小了啊,分组之后,那只是在当前组内的一个M乘N,如果说我们的下游任务是上游任务的整数倍的话,那上游就可以是一个一对下游N的关系了,那一乘以N的话,这个跟M乘N比起来,数量级就大大的减少了。
04:30
所以skill这种方式在实际应用的过程当中还是会比较节省资源的好,接下来呢,我们就在代码里边来做一个具体的测试,我们还是新建一个scla的object,当前我们测的是。Potential。Test。呃,那里边的具体逻辑,我们还是可以先把它抄过来。
05:00
上边我们替换成下划线,引入影视转换。然后接下来我们自然想到了,那呃,就把这里的re balanceance,我们直接改成skill,然后做一个测试不就完了吗?呃,但是我们会发现啊,这种测试其实没有太大的效果,为什么呢?因为主要是针对当前要做一个分组的轮询,那假如说我当前的上游。S任务它只有一个并行子任务,那相当于我当前就只能分一组啊,那那它永远是一对四的这种关系,那当前的risk skill跟rebance就没有任何区别了,所以如果我们想测出risk和reb的区别的话,那应该上游至少要有两个任务,然后下游有四个,这才能做中间的分组啊。哎,所以我们接下来需要把前面的数据源做一个调整,那我们这里边就干脆啊,因为这个click s它本身就是一个非行的s function嘛,所以这里边我们干脆自己再去单独的定义一个并行的数据源好了。
06:07
我们干脆啊,使用匿名类的这种方式来做一个实现吧,不去单独定义了,我们直接实现reach parallel source function里边的数据类型的话,我们简单一点啊,干脆就用整数吧,就12345678,我们就用这样的八个数来进行一个分配。好。那里边必须要去实现的方法,我们知道一个wrong,一个cancel啊,本身这里边的逻辑的话,我们现在不需要随机生成,那就会很简单了,我们直接一个for循环,比方说我们直接啊,哎。从零到七。然后接下来呢,不停的生成,我们我们生成还是一到八的数吧,那就变成了I加一啊,每次发送的数据应该都是I加一,然后接下来呢,我们做一个判断,到底分发到。哪一个并行子任务上呢?诶这个时候我们会发现啊,这里如果要去做分发的话,不就是直接调用ctx它的。
07:07
Collect这样一个方法吗?诶,我们就是要把这个I加一这个数直接发送到下游嘛,那我怎么能判断它到底发送到了哪里呢?诶这就需要使用运行时上下文去做一个判断了。我们这里啊,利用。运行时。上下文中的。Sub task,我们还记得可以获取到当前sub task的ID信息。ID信息。来控制。数据由。哪个?并行子任务。生成。啊,所以当前我们这个SS任务啊,简单来讲之后,我们应该是想要把它的并行度,我们这里实现了这个S之后啊,后边的并行度希望要把它要设置成二。
08:03
那么接下来我们这里就有两个SS任务生成数据源,那我当前的这个一到八八个数到底怎么样去生成呢?诶,那我就控制一下啊,我当前在运行的时候就能获取到当前到底是第零号还是第一号并行子任务。之前我们记得萨task的ID啊,获取的时候是从零开始的,零号和一号,那么这个时候我们就判断了,那如果是当前是零号的话,那我就输出,比方说我就输出偶数吧,2468,输出到第一个这个零号并行子任务上,1357,那就输出到一号并行子任上。所以这个时候呢,只要。在运行至上下文里面获取到这个信息,然后做一个判断,一判断就可以了,所以我们这里的判断标准就是假如说当前我们get runtime contact。我们去获取当前的index of this sub task。假如说它等于。
09:06
当前,哎,那其实就是要看当前这个I到底是奇数还是偶数了,那奇数偶数怎么判断呢?其实就是对二做一个取模运算就可以了。那如果I是偶数的话,那么它对二取模得到的结果是零啊,那我们判断是否跟当前并行子任务的那个序号零和一做一个判断是否相等,相等的话就输出啊,这里需要注意一下是我们前面这个I啊,这里是从零到七,最后输出的是I加一,所以这里边我们判断的也应该是I加一。最后输出的是奇数,那么我们就判断它对二取模要等于一,从第一个并行子任务上去输出,那如果它I加一输出的是偶数的话,那么对二取模式零,那么我们就从ID为零的那个并行子任务上去把它发出去,哎,所以整个的这个逻辑其实就是这样,做一个一判断就可以了。
10:04
啊,然后另外我们还有这个cancel cancel这里,呃,我们其实直接不更改也可以啊,直接用这个问号做一个占位符,相当于什么都不处理啊,相当于是空的,因为我们这里边其实就是只有八条数据发完了就结束了啊,不需要再去做中断了,好有了这一步操作之后,接下来我们就可以运行一下做一个具体的测试了。我们看看现在得到的结果是什么样子。我们可以看到当前输出的结果呢?一眼看上去好像没什么规律,但是如果我们仔细的观察一下的话,就会发现啊一三。五七他们分发到的对应的前面的这一个并行子任务的编号,下游输出的时候,我们的并行度是四嘛,它对应的编号是三和四,所有的奇数对应的都是三和四,而且一和五对应的是三。
11:04
三和七对应的四,所以当前我们1357按照顺序啊,输出的时候,其实就是3434啊,所以对应在我们这张图上,我们就可以看到上边比方说啊,这个是零号并行责任物,这是2468。下边是一号并行子任务,这是1357。SS任务进来之后,接下来下一步任务是一个,哎,我们就直接打印输出了,下游是四个print并行子任务,所以到底在哪一个并行子任务上打印输出呢?1357来的就都是。第三个和第四个啊,都是放到下面来打印输出,那同样对应的是不是2468就应该是在第一个和第二个上呢?确实是这样,我们看到2468它对应的就是1212,在上面的两个当前自己分组内进行一个轮询的分配。
12:04
这就是我们所说的skill啊,当然如果说这里我们把这个更改一下,改成reb的话,就会非常明显的看到和reb的区别。我们运行一下。我们可以看到这个结果跟之前我们做的时候就完全不一样了啊,那它的这个规律又是什么呢?这个规律我们还是要分奇偶数来看啊,因为现在的这个数据源还是分奇偶数在两个不同的并行子任务上去生成的嘛,我们看上边都是偶数,下边都是奇数,那偶数的规律是什么呢?2468,它对应的输出是。2341,哎,我们我们说的下游不是四个并行子任务吗?所以2341,这不就是一个轮询发牌的过程吗?四个任务全部都在轮询,而下面的1357,我们看它输出的顺序是3412啊,其实还是3412,还是四个任务在轮巡发牌,只不过他们是各轮询各的,彼此之间并没有什么关联。
13:10
通过这个例子,我们可以非常直观的看到reb和的区别。
我来说两句