经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Flink » 查看文章
Flink源码分析 - 剖析一个简单的Flink程序
来源：cnblogs　　作者：chaojianok　　时间：2019/6/6 8:34:01　　对本文有异议

本篇文章首发于头条号Flink程序是如何执行的？通过源码来剖析一个简单的Flink程序，欢迎关注头条号和微信公众号“大数据技术和人工智能”（微信搜索bigdata_ai_tech）获取更多干货，也欢迎关注我的CSDN博客。

在这之前已经介绍了如何在本地搭建Flink环境和如何创建Flink应用和如何构建Flink源码，这篇文章用官方提供的SocketWindowWordCount例子来解析一下一个常规Flink程序的每一个基本步骤。

示例程序

  1. public class SocketWindowWordCount {
  2.     public static void main(String[] args) throws Exception {
  3.         // the host and the port to connect to
  4.         final String hostname;
  5.         final int port;
  6.         try {
  7.             final ParameterTool params = ParameterTool.fromArgs(args);
  8.             hostname = params.has("hostname") ? params.get("hostname") : "localhost";
  9.             port = params.getInt("port");
  10.         } catch (Exception e) {
  11.             System.err.println("No port specified. Please run 'SocketWindowWordCount " +
  12.                     "--hostname <hostname> --port <port>', where hostname (localhost by default) " +
  13.                     "and port is the address of the text server");
  14.             System.err.println("To start a simple text server, run 'netcat -l <port>' and " +
  15.                     "type the input text into the command line");
  16.             return;
  17.         }
  18.         // get the execution environment
  19.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         // get input data by connecting to the socket
  21.         DataStream<String> text = env.socketTextStream(hostname, port, "\n");
  22.         // parse the data, group it, window it, and aggregate the counts
  23.         DataStream<WordWithCount> windowCounts = text
  24.                 .flatMap(new FlatMapFunction<String, WordWithCount>() {
  25.                     @Override
  26.                     public void flatMap(String value, Collector<WordWithCount> out) {
  27.                         for (String word : value.split("\\s")) {
  28.                             out.collect(new WordWithCount(word, 1L));
  29.                         }
  30.                     }
  31.                 })
  32.                 .keyBy("word")
  33.                 .timeWindow(Time.seconds(5))
  35.                 .reduce(new ReduceFunction<WordWithCount>() {
  36.                     @Override
  37.                     public WordWithCount reduce(WordWithCount a, WordWithCount b) {
  38.                         return new WordWithCount(a.word, a.count + b.count);
  39.                     }
  40.                 });
  41.         // print the results with a single thread, rather than in parallel
  42.         windowCounts.print().setParallelism(1);
  43.         env.execute("Socket Window WordCount");
  44.     }
  45.     // ------------------------------------------------------------------------
  46.     /**
  47.      * Data type for words with count.
  48.      */
  49.     public static class WordWithCount {
  50.         public String word;
  51.         public long count;
  52.         public WordWithCount() {}
  53.         public WordWithCount(String word, long count) {
  54.             this.word = word;
  55.             this.count = count;
  56.         }
  57.         @Override
  58.         public String toString() {
  59.             return word + " : " + count;
  60.         }
  61.     }
  62. }

上面这个是官网的SocketWindowWordCount程序示例，它首先从命令行中获取socket连接的host和port，然后获取执行环境、从socket连接中读取数据、解析和转换数据，最后输出结果数据。
每个Flink程序都包含以下几个相同的基本部分：

  1. 获得一个execution environment，
  2. 加载/创建初始数据，
  3. 指定此数据的转换，
  4. 指定放置计算结果的位置，
  5. 触发程序执行

Flink执行环境

  1. final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Flink程序都是从这句代码开始，这行代码会返回一个执行环境，表示当前执行程序的上下文。如果程序是独立调用的，则此方法返回一个由createLocalEnvironment()创建的本地执行环境LocalStreamEnvironment。从其源码里可以看出来：

  1. //代码目录：org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
  2. public static StreamExecutionEnvironment getExecutionEnvironment() {
  3.     if (contextEnvironmentFactory != null) {
  4.         return contextEnvironmentFactory.createExecutionEnvironment();
  5.     }
  6.     ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  7.     if (env instanceof ContextEnvironment) {
  8.         return new StreamContextEnvironment((ContextEnvironment) env);
  9.     } else if (env instanceof OptimizerPlanEnvironment || env instanceof PreviewPlanEnvironment) {
  10.         return new StreamPlanEnvironment(env);
  11.     } else {
  12.         return createLocalEnvironment();
  13.     }
  14. }

获取输入数据

  1. DataStream<String> text = env.socketTextStream(hostname, port, "\n");

这个例子里的源数据来自于socket，这里会根据指定的socket配置创建socket连接，然后创建一个新数据流，包含从套接字无限接收的字符串，接收的字符串由系统的默认字符集解码。当socket连接关闭时，数据读取会立即终止。通过查看源码可以发现，这里实际上是通过指定的socket配置来构造一个SocketTextStreamFunction实例，然后源源不断的从socket连接里读取输入的数据创建数据流。

  1. //代码目录：org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
  2. @PublicEvolving
  3. public DataStreamSource<String> socketTextStream(String hostname, int port, String delimiter, long maxRetry) {
  4.     return addSource(new SocketTextStreamFunction(hostname, port, delimiter, maxRetry),
  5.             "Socket Stream");
  6. }

SocketTextStreamFunction的类继承关系如下：
SocketTextStreamFunction类关系图

可以看出SocketTextStreamFunction是SourceFunction的子类，SourceFunction是Flink中所有流数据源的基本接口。SourceFunction的定义如下：

  1. //代码目录：org/apache/flink/streaming/api/functions/source/SourceFunction.java
  2. @Public
  3. public interface SourceFunction<T> extends Function, Serializable {
  4.     void run(SourceContext<T> ctx) throws Exception;
  5.     void cancel();
  6.     @Public
  7.     interface SourceContext<T> {
  8.         void collect(T element);
  9.         @PublicEvolving
  10.         void collectWithTimestamp(T element, long timestamp);
  11.         @PublicEvolving
  12.         void emitWatermark(Watermark mark);
  13.         @PublicEvolving
  14.         void markAsTemporarilyIdle();
  15.         Object getCheckpointLock();
  16.         void close();
  17.     }
  18. }

SourceFunction定义了run和cancel两个方法和SourceContext内部接口。

  • run(SourceContex)：实现数据获取逻辑，并可以通过传入的参数ctx进行向下游节点的数据转发。
  • cancel()：用来取消数据源，一般在run方法中，会存在一个循环来持续产生数据，cancel方法则可以使该循环终止。
  • SourceContext：source函数用于发出元素和可能的watermark的接口，返回source生成的元素的类型。

了解了SourceFunction这个接口，再来看下SocketTextStreamFunction的具体实现（主要是run方法），逻辑就已经很清晰了，就是从指定的hostname和port持续不断的读取数据，按回车换行分隔符划分成一个个字符串，然后再将数据转发到下游。现在回到StreamExecutionEnvironment的socketTextStream方法，它通过调用addSource返回一个DataStreamSource实例。思考一下，例子里的text变量是DataStream类型，为什么源码里的返回类型却是DataStreamSource呢？这是因为DataStream是DataStreamSource的父类，下面的类关系图可以看出来，这也体现出了Java的多态的特性。
DataStreamSource类关系图

数据流操作

对上面取到的DataStreamSource，进行flatMap、keyBy、timeWindow、reduce转换操作。

  1. DataStream<WordWithCount> windowCounts = text
  2.         .flatMap(new FlatMapFunction<String, WordWithCount>() {
  3.             @Override
  4.             public void flatMap(String value, Collector<WordWithCount> out) {
  5.                 for (String word : value.split("\\s")) {
  6.                     out.collect(new WordWithCount(word, 1L));
  7.                 }
  8.             }
  9.         })
  10.         .keyBy("word")
  11.         .timeWindow(Time.seconds(5))
  12.         .reduce(new ReduceFunction<WordWithCount>() {
  13.             @Override
  14.             public WordWithCount reduce(WordWithCount a, WordWithCount b) {
  15.                 return new WordWithCount(a.word, a.count + b.count);
  16.             }
  17.         });

这段逻辑中，对上面取到的DataStreamSource数据流分别做了flatMap、keyBy、timeWindow、reduce四个转换操作，下面说一下flatMap转换，其他三个转换操作读者可以试着自己查看源码理解一下。

先看一下flatMap方法的源码吧，如下。

  1. //代码目录：org/apache/flink/streaming/api/datastream/DataStream.java
  2. public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) {
  3.     TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes(clean(flatMapper),
  4.             getType(), Utils.getCallLocationName(), true);
  5.     return transform("Flat Map", outType, new StreamFlatMap<>(clean(flatMapper)));
  6. }

这里面做了两件事，一是用反射拿到了flatMap算子的输出类型，二是生成了一个operator。flink流式计算的核心概念就是将数据从输入流一个个传递给operator进行链式处理，最后交给输出流的过程。对数据的每一次处理在逻辑上成为一个operator。上面代码中的最后一行transform方法的作用是返回一个SingleOutputStreamOperator，它继承了Datastream类并且定义了一些辅助方法，方便对流的操作。在返回之前，transform方法还把它注册到了执行环境中。下面这张图是一个由Flink程序映射为Streaming Dataflow的示意图：
Flink基本编程模型

结果输出

  1. windowCounts.print().setParallelism(1);

每个Flink程序都是以source开始以sink结尾，这里的print方法就是把计算出来的结果sink标准输出流。在实际开发中，一般会通过官网提供的各种Connectors或者自定义的Connectors把计算好的结果数据sink到指定的地方，比如Kafka、HBase、FileSystem、Elasticsearch等等。这里的setParallelism是设置此接收器的并行度的，值必须大于零。

执行程序

  1. env.execute("Socket Window WordCount");

Flink有远程模式和本地模式两种执行模式，这两种模式有一点不同，这里按本地模式来解析。先看下execute方法的源码，如下：

  1. //代码目录：org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
  2. @Override
  3. public JobExecutionResult execute(String jobName) throws Exception {
  4.     // transform the streaming program into a JobGraph
  5.     StreamGraph streamGraph = getStreamGraph();
  6.     streamGraph.setJobName(jobName);
  7.     JobGraph jobGraph = streamGraph.getJobGraph();
  8.     jobGraph.setAllowQueuedScheduling(true);
  9.     Configuration configuration = new Configuration();
  10.     configuration.addAll(jobGraph.getJobConfiguration());
  11.     configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");
  12.     // add (and override) the settings with what the user defined
  13.     configuration.addAll(this.configuration);
  14.     if (!configuration.contains(RestOptions.BIND_PORT)) {
  15.         configuration.setString(RestOptions.BIND_PORT, "0");
  16.     }
  17.     int numSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());
  18.     MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder()
  19.         .setConfiguration(configuration)
  20.         .setNumSlotsPerTaskManager(numSlotsPerTaskManager)
  21.         .build();
  22.     if (LOG.isInfoEnabled()) {
  23.         LOG.info("Running job on local embedded Flink mini cluster");
  24.     }
  25.     MiniCluster miniCluster = new MiniCluster(cfg);
  26.     try {
  27.         miniCluster.start();
  28.         configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());
  29.         return miniCluster.executeJobBlocking(jobGraph);
  30.     }
  31.     finally {
  32.         transformations.clear();
  33.         miniCluster.close();
  34.     }
  35. }

这个方法包含三部分：将流程序转换为JobGraph、使用用户定义的内容添加（或覆盖）设置、启动一个miniCluster并执行任务。关于JobGraph暂先不讲，这里就只说一下执行任务，跟进下return miniCluster.executeJobBlocking(jobGraph);这行的源码，如下：

  1. //代码目录：org/apache/flink/runtime/minicluster/MiniCluster.java
  2. @Override
  3. public JobExecutionResult executeJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
  4.     checkNotNull(job, "job is null");
  5.     final CompletableFuture<JobSubmissionResult> submissionFuture = submitJob(job);
  6.     final CompletableFuture<JobResult> jobResultFuture = submissionFuture.thenCompose(
  7.         (JobSubmissionResult ignored) -> requestJobResult(job.getJobID()));
  8.     final JobResult jobResult;
  9.     try {
  10.         jobResult = jobResultFuture.get();
  11.     } catch (ExecutionException e) {
  12.         throw new JobExecutionException(job.getJobID(), "Could not retrieve JobResult.", ExceptionUtils.stripExecutionException(e);
  13.     }
  14.     try {
  15.         return jobResult.toJobExecutionResult(Thread.currentThread().getContextClassLoader());
  16.     } catch (IOException | ClassNotFoundException e) {
  17.         throw new JobExecutionException(job.getJobID(), e);
  18.     }
  19. }

这段代码的核心逻辑就是final CompletableFuture<JobSubmissionResult> submissionFuture = submitJob(job);，调用了MiniCluster类的submitJob方法，接着看这个方法：

  1. //代码目录：org/apache/flink/runtime/minicluster/MiniCluster.java
  2. public CompletableFuture<JobSubmissionResult> submitJob(JobGraph jobGraph) {
  3.     final CompletableFuture<DispatcherGateway> dispatcherGatewayFuture = getDispatcherGatewayFuture();
  4.     // we have to allow queued scheduling in Flip-6 mode because we need to request slots
  5.     // from the ResourceManager
  6.     jobGraph.setAllowQueuedScheduling(true);
  7.     final CompletableFuture<InetSocketAddress> blobServerAddressFuture = createBlobServerAddress(dispatcherGatewayFuture);
  8.     final CompletableFuture<Void> jarUploadFuture = uploadAndSetJobFiles(blobServerAddressFuture, jobGraph);
  9.     final CompletableFuture<Acknowledge> acknowledgeCompletableFuture = jarUploadFuture
  10.         .thenCombine(
  11.             dispatcherGatewayFuture,
  12.             (Void ack, DispatcherGateway dispatcherGateway) -> dispatcherGateway.submitJob(jobGraph, rpcTimeout))
  13.         .thenCompose(Function.identity());
  14.     return acknowledgeCompletableFuture.thenApply(
  15.         (Acknowledge ignored) -> new JobSubmissionResult(jobGraph.getJobID()));
  16. }

这里的Dispatcher组件负责接收作业提交，持久化它们，生成JobManagers来执行作业并在主机故障时恢复它们。Dispatcher有两个实现，在本地环境下启动的是MiniDispatcher，在集群环境上启动的是StandaloneDispatcher。下面是类结构图：
MiniDispatcher类结构图

这里的Dispatcher启动了一个JobManagerRunner，委托JobManagerRunner去启动该Job的JobMaster。对应的代码如下：

  1. //代码目录：org/apache/flink/runtime/jobmaster/JobManagerRunner.java
  2. private CompletableFuture<Void> verifyJobSchedulingStatusAndStartJobManager(UUID leaderSessionId) {
  3.     final CompletableFuture<JobSchedulingStatus> jobSchedulingStatusFuture = getJobSchedulingStatus();
  4.     return jobSchedulingStatusFuture.thenCompose(
  5.         jobSchedulingStatus -> {
  6.             if (jobSchedulingStatus == JobSchedulingStatus.DONE) {
  7.                 return jobAlreadyDone();
  8.             } else {
  9.                 return startJobMaster(leaderSessionId);
  10.             }
  11.         });
  12. }

JobMaster经过一系列方法嵌套调用之后，最终执行到下面这段逻辑：

  1. //代码目录：org/apache/flink/runtime/jobmaster/JobMaster.java
  2. private void scheduleExecutionGraph() {
  3.     checkState(jobStatusListener == null);
  4.     // register self as job status change listener
  5.     jobStatusListener = new JobManagerJobStatusListener();
  6.     executionGraph.registerJobStatusListener(jobStatusListener);
  7.     try {
  8.         executionGraph.scheduleForExecution();
  9.     }
  10.     catch (Throwable t) {
  11.         executionGraph.failGlobal(t);
  12.     }
  13. }

这里executionGraph.scheduleForExecution();调用了ExecutionGraph的启动方法。在Flink的图结构中，ExecutionGraph是真正被执行的地方，所以到这里为止，一个任务从提交到真正执行的流程就结束了，下面再回顾一下本地环境下的执行流程：

  1. 客户端执行execute方法；
  2. MiniCluster完成了大部分任务后把任务直接委派给MiniDispatcher；
  3. Dispatcher接收job之后，会实例化一个JobManagerRunner，然后用这个实例启动job；
  4. JobManagerRunner接下来把job交给JobMaster去处理；
  5. JobMaster使用ExecutionGraph的方法启动整个执行图，整个任务就启动起来了。

原文链接：http://www.cnblogs.com/cjblogs/p/10978975.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Flink热门文章

  • 深度介绍Flink在字节跳动数据流的实践
  • idea中flink启动报错org.apache.flink.api.common.ExecutionConfig$GlobalJobParameters
  • flink-cdc同步mysql数据到elasticsearch
  • 实时计算框架：Flink集群搭建与运行机制
  • Flink的Job启动Driver端(源码分析)
  • Flink中TaskManager端执行用户逻辑过程(源码分析)
  • 统一批处理流处理——Flink批流一体实现原理
  • Flink+Druid构建实时OLAP的探索
  • 01.Flink笔记-编译、部署
  • 一文解开主流开源变更数据捕获技术之Flink CDC的入门使用

    • Flink推荐文章

  • Flink Batch Hash Aggregate
  • flink-cdc同步mysql数据到elasticsearch
  • 一文解开主流开源变更数据捕获技术之Flink CDC的入门使用
  • 实例讲解Flink 流处理程序编程模型
  • 《关于我因为flink成为spark源码贡献者这件小事》
  • Flink SQL管理平台flink-streaming-platform-web安装搭建
  • 深度介绍Flink在字节跳动数据流的实践
  • 实时计算框架：Flink集群搭建与运行机制
  • Flink-3-ApacheFlink架构
  • Flink-5-DataStreamAPI

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号