05 Flink 的 WordCount

  • 开发
  • 47

前言

本文对应于 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 内容如下 

a0b994935dfb439a9500092dce659245.png

 

 

整体交互流程 

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

4488db13d6934b61b44b01fdd687c756.png

然后 TaskManager 和 JobManager 这边交互如下 

cde2d6cfdfbe49448caebb3e0219759e.png 

 

Driver 这边的处理  

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

a3e0f50291034d6d900d9a668aec3c62.png

 

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

801f3e46704d40aeb79d1dc04c32e660.png 

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

c2c7661153954cf3a8a4cbb62a4e7b0f.png 

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

ca22a3db7d8c40d8b1155722c59b28d3.png 

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

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

db20124bd6cd419b9a8b0b9a4d302d64.png

 

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

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

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

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

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

244ec2a2627e4c92abc08a542bb67e0c.png

 

 

JobManager 这边的处理 

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

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

然后就是向 Dispatcher 提交 jobGraph

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

a727614e3b2c49f7a1fdfa267dab09a9.png

 

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

450ee6fb0ab04a34808411d478dd51fc.png 

然后是 JobManagerRunner 启动 JobMaster

ff9c69b8b0504b279343dff237ae828a.png

 

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

2dc84b523fd34aa0815b323dd907dd7b.png

 

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

28daa5ab7488408fb96cf3ba5b5da170.png

 

 

 TaskManager 这边的处理

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

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

3298bcfb67ed4f55809691b51527a2c0.png

 

然后就是 Task 这边的执行 

945e20bee9f041fa99207604df6d8239.png 

这边执行的就是基于 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的时候, 在迭代记录将记录输出到下游 

495e543574d145fc93cf59704922f9e4.png

 

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

2b2c1476c3f84c6c8c8aaec7778d27d5.png 

repartition 的地方 

b3a65c985ba1472e83268605a15c5756.png 

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

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

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

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

ef806d3fc228449aabe980c3c34f3241.png

 

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

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

6812aa0d6c2843f18bdf651390caf66f.png

 

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

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

2cb90acbfbe144d0bcddfc31787829cb.png

 

 

各个 SubTask 的数据交互

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

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

a1489d547145440e845e525d200c8595.png

 

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

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

52c1f8dc2c9b49b1a42bf81f5f8ee621.png

 

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

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

131e23b40835406d8674334574beffac.png

 

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

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

d931bbcb2aff4e708703f8a0f6ec7675.png

 

然后我们看一下 下游的 SubTask 这边来消费上游的 SubTask 的处理, 这里获取的是 InputChannel 的 partitionId

ed6b6ac662fb486a86b97ffd26ab9e84.png 

InputChannel 的 partitionId 是来自于上游的 inputChannelDescriptor.partitionId, 这样就把整个流程串联起来了 

下游的 SubTask 可以读取到所有上游SubTask 的结果信息 

2899588ab8c84074a34257a68b1a9e95.png

 

 

各个 Task 执行的通知 

Task 和 Task 之间是有依赖关系的, 下游的 Task 相对于 上游的Task 称之为 consumers

当上游的 Task 有数据提交到之后, 这里会通知到 JobMaster 通知 partitionId 已经在产出数据了 

然后 jobMaster 通知该 Task 的下游的 Task 开始执行 

d68d2b90be3a4d2aa39ad5846508af00.png

 

然后 JobMaster 这边收到 scheduleOrUpdateConsumers 之后的处理如下 

开始 调度下游的 consumers, 即下游的 SubTask 开始申请资源, 然后 执行 等等

6fdc0b75e85e478597b8c60771b2ef91.png

2da373267e9f442da186f19fed875c2f.png 

 

计算结果的交互 

首先是 driver 这边 

Job 提交了之后, 会执行 requestJobResult, 这里面是向 JobManager 这边发送 http 请求, 获取 给定的 Job 的执行结果 

8ef34abcbe5c40d9a085e32148669366.png

 

发送的 http 请求如下, 请求的是 “/v1/jobs/682debfc2ac22e73847c23b1953343e1/execution-result”

d03f858b568d43c6aeba184ddbffb14e.png 

然后 JobMaster 这边的处理如下, 我们这里 需要关注的是 这个 accumulatorResults, 这里面暂存的我们计算的结果 

5b385e2b66714324b34ca16e69fcb1e3.png 

然后这个 accumulatorResults 的数据来自于各个 Task 执行完成之后通知到 JobMaster 这边的 accumulators 

如下图 这里是 Task 执行完成之后提交更新任务执行状态的请求到 JobMaster, state 中携带了 accumulators

07c14a7d69e149f9960a1f68b7cdcdd7.png

 

接下来是封装 ArchivedExecutionGraph, 这里封装的 accumulators 是使用的各个 Task 执行完成之后响应的 accumulators

2b8227576bf74c7f98410f769258836c.png

 

然后 executionGraph.getAccumulatorsSerialized 遍历的基础 accumulatos 是来自于如下, 可以看到的是遍历的当前执行计划的所有的 Vertex 的 accumulators

然后 结合上上一张图可以看到的是 这里是从 vertex.attempt 中获取的数据, 然后 vertex.attempt 的数据是来自于 ExecutionGraph.updateStateInternal

然后外层 JobMaster.jobStatusChanged 将这上面生成的 ArchivedExecutionGraph 设置到了 JobManagerRunner.resultFuture 中 

81fc8e086dc64fecbefbd61c137fde9d.png

 

Task 这边, 任务执行完成之后, 将 accumulators 封装到 TaskExecutionState, 然后响应给 JobMaster

e2551d79a7dc434ba9c7dd63c8672e63.png 

 

完 

 

 

 

阅读全部

相关推荐

  2. spark与flink的wordcount示例

    2024-02-22 08:50:04       33 阅读

  4. flink 入门学习 wordcount

    2024-02-22 08:50:04       30 阅读

  5. 第一个FLink程序之wordCount

    2024-02-22 08:50:04       54 阅读

  7. 【Flink on k8s】- 7 - 在本地运行第一个 flink wordcount job

    2024-02-22 08:50:04       54 阅读

最近更新

  3. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-02-22 08:50:04       94 阅读

  4. Could not load dynamic library ‘cudart64_100.dll‘

    2024-02-22 08:50:04       101 阅读

  9. 在Django里面运行非项目文件

    2024-02-22 08:50:04       82 阅读

  19. Python语言-面向对象

    2024-02-22 08:50:04       91 阅读

  20. 如何分清楚常见的 Git 分支管理策略Git Flow、GitHub Flow 和 GitLab Flow

    2024-02-22 08:50:04       85 阅读

热门阅读

  3. Python游戏开发：Pygame框架与实例

    2024-02-22 08:50:04       45 阅读

  4. Arm的Cortex-A和Cortex-X内核架构对比

    2024-02-22 08:50:04       46 阅读

  5. 数据结构：入门概述

    2024-02-22 08:50:04       52 阅读

  9. Git 备忘清单

    2024-02-22 08:50:04       49 阅读

  10. 好奇！为什么gateway和springMVC之间依赖冲突？

    2024-02-22 08:50:04       55 阅读

  11. origin/master master

    2024-02-22 08:50:04       50 阅读

  12. ThinkPHP6中使用GatewayWorker

    2024-02-22 08:50:04       54 阅读

  13. 论文选题分享及思路（一）《基于C51单片机的自动化测量产线的设计》

    2024-02-22 08:50:04       41 阅读

  14. Nginx安装Lua

    2024-02-22 08:50:04       50 阅读

  20. 【职场经验】关于自动化用例设计的思考

    2024-02-22 08:50:04       55 阅读

  27. 关于大语言模型LLM相关的数据集、预训练模型、提示词、微调的文心一言问答

    2024-02-22 08:50:04       60 阅读

  28. 设计模式-建造者模式

    2024-02-22 08:50:04       48 阅读