从flink源码中案例开始讲解

  1. public class WordCount {
  2.     public static void main(String[] args) throws Exception {
  4.         final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
  6.         // set up the execution environment
  7.         final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  9.         // make parameters available in the web interface
  10.         env.getConfig().setGlobalJobParameters(params);
  12.         // get input data
  13.         DataSet<String> text = null;
  14.         if (params.has("input")) {
  15.             // union all the inputs from text files
  16.             for (String input : params.getMultiParameterRequired("input")) {
  17.                 if (text == null) {
  18.                     text = env.readTextFile(input);
  19.                 } else {
  20.                     text = text.union(env.readTextFile(input));
  21.                 }
  22.             }
  23.             Preconditions.checkNotNull(text, "Input DataSet should not be null.");
  24.         } else {
  25.             // get default test text data
  26.             System.out.println("Executing WordCount example with default input data set.");
  27.             System.out.println("Use --input to specify file input.");
  28.             text = WordCountData.getDefaultTextLineDataSet(env);
  29.         }
  31.         DataSet<Tuple2<String, Integer>> counts =
  32.                 // split up the lines in pairs (2-tuples) containing: (word,1)
  33.                 text.flatMap(new Tokenizer())
  34.                 // group by the tuple field "0" and sum up tuple field "1"
  35.                 .groupBy(0)
  36.                 .sum(1);
  38.         // emit result
  39.         if (params.has("output")) {
  40.             counts.writeAsCsv(params.get("output"), "\n", " ");
  41.             // execute program
  42.             env.execute("WordCount Example");
  43.         } else {
  44.             System.out.println("Printing result to stdout. Use --output to specify output path.");
  45.             counts.print();
  46.         }
  48.     }

程序的最终开始执行是从 env.execute开始
image.png
从 env.execute 点击 ，我们首先看 final JobClient jobClient = executeAsync(jobName); 从这个方法继续点击
image.png
从这个方法中，我们可以看到程序开始执行的位置
image.png
从这个方法中查看详细的执行步骤
image.png
可以看出，所有的执行器是 PipelineExecutor 接口， 具体的执行器，需要继承该方法
image.png
该LocalExecutor的** execute 方法中，可以看到详细的作业执行流程，接下来，我们分别从这几个方法中入手，详细分析具体的实现步骤细节**
image.png

1.getJobGraph 方法中详细 将 streamGraph 如何转化为 JobGraph
2.startMiniCluster 方法是如何 启动ResourceManager， JobMaster 等 信息
3.submitJob如何提交自己写的代码作业