当前位置：网站首页>Flink入门到实战-阶段二(图解运行时架构)

Flink入门到实战-阶段二(图解运行时架构)

2022-07-17 05:24:00 【顶尖高手养成计划】

系统架构

提交作业流程

高级抽象视角

独立模式

Yarn集群

会话模式

1.先对于yarn申请一个JobManager

2.JobManager处理任务

单作业模式

数据流程图

所有的 Flink 程序都可以归纳为由三部分构成：Source、Transformation 和 Sink。

Source 表示“源算子”，负责读取数据源。

Transformation 表示“转换算子”，利用各种算子进行处理加工。

Sink 表示“下沉算子”，负责数据的输出。

并行度

流程图

实验

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很像

执行图

在这个转换过程中，有几个不同的阶段，会生成不同层级的图，其中最重要的就是作业图

（JobGraph）和执行图（ExecutionGraph）。Flink 中任务调度执行的图，按照生成顺序可以分成

四层：

逻辑流图（StreamGraph）→ 作业图（JobGraph）→ 执行图（ExecutionGraph）→ 物理

图（Physical Graph）。

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

理论

任务槽（Task Slots）

之前已经提到过，Flink 中每一个 worker(也就是 TaskManager)都是一个 JVM 进程，它可以启动多个独立的线程，来并行执行多个子任务（subtask）。所以如果想要执行 5 个任务，并不一定非要 5 个 TaskManager，我们可以让 TaskManager多线程执行任务。如果可以同时运行 5 个线程，那么只要一个 TaskManager 就可以满足我们之

前程序的运行需求了。

任务对任务槽的共享

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

实验

前期准备

现在的情况是有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();
    }
}