System architecture

Submit the workflow  

Advanced Abstract Perspective

Independent mode  

Yarn colony  

Conversational mode

1. First for yarn Apply for one JobManager

 2.JobManager Processing tasks

Single operation mode  

Data flow chart

Parallelism  

flow chart

experiment

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        // Get the execution environment 
        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);

        // For the stream data of the resulting tuple , Group aggregation 
        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();

        // Because it is a stream handler , So here we need to constantly implement 
        env.execute();
    }
}

The result

Conclusion  

• It can be seen that Flink Parallelism and Spark The concept of repartition is very similar to

Chain of operators

experiment  

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        // Get the execution environment 
        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);
            }
        });

        // For the stream data of the resulting tuple , Group aggregation 
        map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).print();

        // Because it is a stream handler , So here we need to constantly implement 
        env.execute();
    }
}

result  

Conclusion  

• You can see the operators of one-to-one relationship. They execute in a task
• If it's one to many , Then it is to separate two tasks
• This and Spark Inside Shuffle Split Stage It's like  

Execution diagram

Mission （Tasks） And task slots （Task Slots）

theory

Task to task slot sharing

The following is an example of allocation Relationship between task slot and parallelism

experiment

Preparation

The current situation is 3 individual task slots, If we set the parallelism to 5 What's the situation

Experimental code  

public class FlinkSoctet {
    public static void main(String[] args) throws Exception {
        // Get the execution environment 
        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);
            }
        });

        // For the stream data of the resulting tuple , Group aggregation 
        map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).print();

        // Because it is a stream handler , So here we need to constantly implement 
        env.execute();
    }
}

result

  Report errors

Task scheduling

According to your own requirements , To distribute reasonably task In different taskslots, Understanding can , because taskslots Sharing is an optimization of task execution