说明
所有上游数据下发到下游 第一个分区
提示： 在 kafka 环境下，需要创建多个分区
案例
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 F10global_kafka {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> dss = env.fromElements("lilei", "lili", "lucy", "hanmeimei", "lilei", "lili", "lucy", "hanmeimei");
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);
}
})
.setParallelism(4); // 设置并行度，与上游算子并行度不同，将数据轮询下发到下游算子（rebalance）
mapDS.print("mapDS::");
//
DataStream<Tuple2<Integer,String>> globalDS = dss
.global()// 发送到第一个分区
.map(new RichMapFunction<String, Tuple2<Integer,String>>() {
@Override
public Tuple2<Integer, String> map(String s) throws Exception {
return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(),s);
}
})
.setParallelism(4); // 设置并行度，与上游算子并行度不同，由于上面调用 global() ，将数据发送到第一个分区
globalDS.print("globalDS::");
System.out.println(env.getExecutionPlan());
env.execute();
}
}
提示：设置并行度的原因：上游算子与下游算子并行度不同，可以看到是将数据 轮询 下发到下游算子（rebalance）
生成执行计划图
访问：https://flink.apache.org/visualizer/
执行结果
mapDS:::1> (0,hanmeimei)
mapDS:::8> (0,hanmeimei)
mapDS:::8> (1,lilei)
mapDS:::7> (1,lilei)
mapDS:::3> (2,lili)
mapDS:::7> (3,lucy)
mapDS:::4> (2,lili)
globalDS:::1> (0,hanmeimei)
globalDS:::3> (0,lili)
globalDS:::4> (0,lucy)
globalDS:::2> (0,lilei)
globalDS:::6> (0,lilei)
globalDS:::7> (0,lili)
mapDS:::8> (3,lucy)
globalDS:::8> (0,lucy)
globalDS:::5> (0,hanmeimei)
可以明显看到：
mapDS打印的数据，是在 4个分区中globalDS打印的数据，是在 第1个分区中
参考
https://blog.51cto.com/u_9928699/5670559