shuffle()

shuffle()，洗牌，是最简单的重分区方式

通过调用 DataStream 的.shuffle()方法，将数据 随机均匀 分配到下游算子

提示：对于同样的输入数据, 每次执行得到的结果不会相同

案例

package chongfenqu;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class F13partitionCustom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置全局并行度是 3
        env.setParallelism(3);

        DataStreamSource<String> dss = env.fromElements("lilei", "lili", "lucy", "hanmeimei","java","flink");

        // 普通算子，与下面做对比
        DataStream<Tuple2<Integer,String>> mapDS = dss.map(new RichMapFunction<String, Tuple2<Integer,String>>() {
            @Override
            public Tuple2<Integer, String> map(String s) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(),s);
            }
        });

        mapDS.print("mapDS::");

        // 随机均匀分配
        DataStream<Tuple2<Integer,String>> shuffleDS = dss
                .shuffle()// 发送到第一个分区
                .map(new RichMapFunction<String, Tuple2<Integer,String>>() {
            @Override
            public Tuple2<Integer, String> map(String s) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(),s);
            }
        });

        shuffleDS.print("shuffleDS::");

        System.out.println(env.getExecutionPlan());
        env.execute();
    }
}

生成可视化图

访问：https://flink.apache.org/visualizer/

执行结果

shuffleDS:::3> (2,lili)
mapDS:::1> (0,lucy)
shuffleDS:::2> (1,lilei)
mapDS:::2> (1,lilei)
mapDS:::3> (2,lili)
mapDS:::2> (1,hanmeimei)
mapDS:::3> (2,java)
shuffleDS:::2> (1,java)
mapDS:::1> (0,flink)
shuffleDS:::3> (2,lucy)
shuffleDS:::2> (1,flink)
shuffleDS:::3> (2,hanmeimei)

说明：

全局并行度是 3

数据是："lilei", "lili", "lucy", "hanmeimei","java","flink"

mapDS 普通算子： rebalance 轮询方式将数据发给分区

第1个数据 lilei 发给 0号分区
第2个数据 lili 发给 1号分区
第3个数据 lucy 发给 0号分区
第4个数据 hanmeimei 发给 1号分区
以此类推

shuffleDS 随机算子： shuffle 随机均匀方式将数据发给分区

第1个数据 lilei 随机发给 1号分区
第2个数据 lili 随机发给 2号分区
第3个数据 lucy 随机发给 2号分区
第4个数据 hanmeimei 随机发给 2号分区
以此类推

rescale()

rescale() 的上下游交互取决于他们的 并行度：

上游为2下游为4，则一个上游对应两个下游
上游为4下游为2，则两个上游对应一个下游
如若是不同倍数的并行度，则下游会有不同数量的输入。

与 `rebalance` 的区别

轮询从下游第一个分区开始以及是点对点分发模式。
rescale 可以增加数据本地处理，减少了网络io性能更高，但数据均衡性不如 rebalance

例子

任务计划

代码

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class Std7Rescale {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Long> ds = env.fromSequence(1,10);
        ds
                .map(new MapFunImpl())
                .print("ds----");
        // 转换操作
        ds
        .rescale()
        .map( new MapFunImpl() )
        .setParallelism(4)
        .print("rescale===");

        System.out.println(env.getExecutionPlan());
        env.execute("my_word_count");
    }

    static class MapFunImpl extends RichMapFunction<Long, Tuple2<Long,Integer>> {

        @Override
        public Tuple2<Long, Integer> map(Long value) throws Exception {
            int index = getRuntimeContext().getIndexOfThisSubtask();
            // 重新组装数据， (原始数据,分区索引)
            return Tuple2.of(value,index);
        }
    }
}

执行结果如下：