05 Flink 的 WordCount

前言

本文对应于 spark 系列的 Spark 的 WordCount

这里主要是从宏观上面来看一下 flink 这边的几个角色, 以及其调度的整个流程

一个宏观大局上的任务的处理, 执行

基于一个本地的 flink 集群

测试用例

/**
 * com.hx.test.Test01WordCount
 *
 * @author Jerry.X.He
 * @version 1.0
 * @date 2021/4/12 10:14
 */
public class Test01WordCount {

    // com.hx.test.Test01WordCount
    // -Xmx100M -XX:+UseSerialGC -XX:+TraceClassUnloading
    public static void main(String[] args) throws Exception {

//        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String jarPath = "D:\\IdeaWorkStations\\HelloFlink\\target\\HelloFlink-0.0.1.jar";
        ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("127.0.0.1", 8081, jarPath);
        env.setParallelism(1);

        String inputPath = "D:\\IdeaWorkStations\\HelloFlink\\src\\main\\resources\\Test01WordCount.txt";
        DataSource<String> inputDs = env.readTextFile(inputPath);

        DataSet<Tuple2<String, Integer>> result = inputDs
                .flatMap(new MyFlatMapMapper())
                .map(new MyMapMapper())
                .groupBy(0)
                .sum(1);
        result.print();

        System.gc();
        System.in.read();
        System.out.println(" end ");

    }

    /**
     * MyFlatMapMapper
     *
     * @author Jerry.X.He
     * @version 1.0
     * @date 2021/4/12 10:24
     */
    private static class MyFlatMapMapper implements FlatMapFunction<String, String> {
        private static List<byte[]> dummyBytes = new ArrayList<>();

        static {
            try {
                for (int i = 0; i < 10; i++) {
                    byte[] tmpBytes = FileUtils.readAllBytes(Paths.get("D:\\IdeaWorkStations\\HelloFlink\\target\\logs\\ROOT.2021-12-27-9.log"));
                    dummyBytes.add(tmpBytes);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                new Thread(new MyRunnable()).start();
            }
        }

        @Override
        public void flatMap(String line, Collector<String> out) throws Exception {
            String[] splits = line.split("\\s+");
            for (String split : splits) {
                out.collect(split);
            }
        }
    }

    /**
     * MyRunnable
     *
     * @author Jerry.X.He
     * @return
     * @date 2021/12/27 16:16
     */
    public static class MyRunnable implements Runnable {
        @Override
        public void run() {
            System.err.println(" MyRunnable.run before ");
            IoUtils.sleep(1000_000);
            System.err.println(" MyRunnable.run after ");
        }
    }

    /**
     * MyMapMapper
     *
     * @author Jerry.X.He
     * @version 1.0
     * @date 2021/4/12 10:29
     */
    private static class MyMapMapper implements MapFunction<String, Tuple2<String, Integer>> {
        @Override
        public Tuple2<String, Integer> map(String word) throws Exception {
            return new Tuple2<>(word, 1);
        }
    }

}

Test01WordCount.txt 内容如下

整体交互流程

Driver 提交 Job 到 JobManager, JobManager 分配任务到 TaskManager

然后 TaskManager 和 JobManager 这边交互如下

Driver 这边的处理

这里是 driver 这边的根据 DataSet阶段转换为 Plan阶段

这里是 Plan阶段转换为 OptimizedPlan阶段

这里是 OptimizedPlan阶段转换为 JobGraph阶段

然后提交的就是 JobGraph, 然后等待集群响应结果信息

这里是将 JobGraph 序列化为为一个临时文件, 然后提交给 flink 集群

然后另外就是 job 这边需要使用的 jar 列表, 也需要提交给集群

然后更详细的提交的请求内容如下, 合计传送了 47kb

然后传递的主要内容为三部分

第一部分为元数据, 内容为 “{"jobGraphFileName": "flink-jobgraph126256148600228610.bin", "jobJarFileNames": ["HelloFlink-0.0.1.jar"], "jobArtifactFileNames": []}”

第二部分为 jobGraph 序列化之后的临时文件, “flink-jobgraph126256148600228610.bin”

第三部分为 job 执行需要的 jar 包, “HelloFlink-0.0.1.jar”

JobManager 这边的处理

JobManager 这边拿到如上 driver 这边提交的 HttpRequest 之后, 处理如下

根据 jobGraphFileName 反序列化 JobGraph, 上传 job 所需要的 jar 到 BlobServer

然后就是向 Dispatcher 提交 jobGraph

返回当前提交的 job 的相关信息, 主要是 jobId

然后是 Dispatcher 这边 persistAndRunJob, 创建 JobManagerRunner, JobMaster

然后是 JobManagerRunner 启动 JobMaster

然后是 JobMaster 这边基于内部的 scheudlerNG 来开始调度任务

然后是根据 ExecutionVertex 封装 TaskDeploymentDescriptor, 然后向 TaskManager 执行 job 拆解之后的任务

TaskManager 这边的处理

TaskExecutor 这边收到了 TaskDeploymentDescriptor 之后, 反序列化 jobInformation, taskInformation, 创建 Task, 然后执行 Task

这里可以从 taskInformation 的上下文信息, 可以看到当前 Task 属于哪一个 JobVertex, 以及改 JobVertex 总共有一个 SubTask, 以及当前 Task 是属于第几个 SubTask

然后就是 Task 这边的执行

这边执行的就是基于 jar 包和 DataSourceTask, ChainnedFlatMapDriver, ChainnedMapDriver 等等来执行具体的业务处理

业务这边执行流程如下

DataSource 逐行读取 Test01WordCount.txt 中的字符串信息, 这里读取的是第一行 “123 234 456”

ChainedFlatMapDriver 这边是将一行的字符串转换为多行字符串, “123 234 456” 转换为 “123”, “234”, “456”

ChainedMapDriver 这边是将一个输入进行映射, 这里的是 “123” 转换为 (“123”, 1)

SynchronousChainedCombineDriver 这边是将到目前为止的结果, hash partition 之后输出到下游的 SubTask

SynchronousChainedCombineDriver 这边先是将记录写入到 sorter, 然后 close的时候, 在迭代记录将记录输出到下游

输出当前 Task 的各个记录的地方

repartition 的地方

每一个 partition 的数据是写出到 RecordWriter 下面的 ReleaseOnSonsumptionResultPartition 下面的 subpartitions[partitionIndex]

然后这部分 ResultPartition 数据是维护在 ResultPartitionManager, 这个是每一个 TaskExecutor 维护一个 ResultPartitionManager 用于相同的任务之间的不同的 SubTask 的数据交互

所以说一个 SubTask 维护了一个 ReleaseOnSonsumptionResultPartition, 然后维护了 parallelism 个 subpartitions

然后下游的 SubTask 依次来遍历上游的 SubTask, 获取当前 SubTask 需要读取的 subpartitions[index] 来作为输入

输出一条记录之后信息如下, bufferBuilder 中输出了 12 字节, 前面四个字节为长度标记 0x04 长度为 4

然后接下来 8个字节为 ”0431323300000001”, 表示的是 “123” + 1, 最前面的 0x04 表示 0x03 + 1[参见 StringValue.writeString]

然后下游的 SubTask 这边读取的是上游的 N 个 SubTask 中的当前 subpartitionIndex 部分的输出, 这里的 partitionId 为上游输入 SubTask 的 partitionId

这里当前 SubTask 会对应 parallelism 个 InputChannel, 每一个分别关联上上游 SubTask 的输出

各个 SubTask 的数据交互

上游任务启动的时候, 会向 PartitionManager 注册输出的 ResultPartition 的信息

然后这里的 partition.getPartitionId 是来自于当前 SubTask 的 partition, 每一个 SubTask 单独生成一个

在 TaskManager 这边传递过程如下, TaskDeploymentDescriptor.producedPartitions.shuffleDescriptor.resultPartitionID -> ReleaseOnConsumptionResultPartition -> ResultPartition

然后再到后面的 ResultPartition.setup 的向 PartitionManager 注册

然后 JobManager 这边是生成这个 shuffleDescriptor 相关, 传递流程如下

IntermediateResultPartition -> Vertex.resultPartitions -> Partition -> PartitionDescriptor -> NettyShuffleDescriptor

然后 Vertex.resultPartitions 这边初始化如下, 里面的组合了一个 IntermediateResultPartition, 它的 partitionId 是传递到后面 NettyShuffleDescriptor 的 partitionID, 然后这个 partitionId 是随机生成的

根据上下文来看, 就是每一个 SubTask 一个