系统架构

提交作业流程 

高级抽象视角

独立模式 

Yarn集群 

会话模式

1.先对于yarn申请一个JobManager

 2.JobManager处理任务

单作业模式 

数据流程图

并行度 

流程图

实验

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        //得到执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> initData = env.socketTextStream("master",9997);

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = initData.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String item, Collector<String> out) throws Exception {
                String[] resItem = item.split(" ");
                for (String s : resItem) {
                    out.collect(s);
                }
            }
        }).setParallelism(1).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String item) throws Exception {
                return Tuple2.of(item, 1);
            }
        }).setParallelism(2);

        //对于得到的元组的流数据，进行分组聚合
        map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).setParallelism(3).print();

        //由于是流处理程序，所以这里要不断的执行
        env.execute();
    }
}

得到的结果

结论 

• 可以看出Flink的并行度和Spark的重分区的概念很像

算子链

实验 

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        //得到执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> initData = env.socketTextStream("master",9997);
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = initData.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String item, Collector<String> out) throws Exception {
                String[] resItem = item.split(" ");
                for (String s : resItem) {
                    out.collect(s);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String item) throws Exception {
                return Tuple2.of(item, 1);
            }
        });

        //对于得到的元组的流数据，进行分组聚合
        map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).print();

        //由于是流处理程序，所以这里要不断的执行
        env.execute();
    }
}

结果 

结论 

• 可以看到一对一的关系的算子他们在一个任务里面执行
• 如果是一对多，那么就是分开两个任务执行
• 这个和Spark里面发生Shuffle拆分Stage很像 

执行图

任务（Tasks）和任务槽（Task Slots）

理论

任务对任务槽的共享

下面是分配的例子任务槽和并行度的关系

实验

前期准备

现在的情况是有3个task slots，如果我们设置并行度为5是什么情况

实验代码 

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        //得到执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> initData = env.socketTextStream("master",9997);
        env.setParallelism(5);

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = initData.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String item, Collector<String> out) throws Exception {
                String[] resItem = item.split(" ");
                for (String s : resItem) {
                    out.collect(s);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String item) throws Exception {
                return Tuple2.of(item, 1);
            }
        });

        //对于得到的元组的流数据，进行分组聚合
        map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).print();

        //由于是流处理程序，所以这里要不断的执行
        env.execute();
    }
}

结果

 报错

任务调度

可以根据自己的要求，合理分配task在不同的taskslots,了解即可，因为taskslots共享就是一种对于任务执行的优化