Flink执行图的生成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

// nc -lk 8888
public void go() throws Exception {
	// 1.准备环境-env
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
	 env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//         env.setRuntimeMode(RuntimeExecutionMode.BATCH);
	//默认是RuntimeExecutionMode.STREAMING
	env.setParallelism(1);

	// 2.准备数据-source
	DataStream<String> linesDS =
			env.fromElements("flink,hadoop,hive", "flink,hadoop,hive", "flink,hadoop", "flink");

	// 3.处理数据-transformation
	DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
		@Override
		public void flatMap(String value, Collector<String> out) throws Exception {
			// value就是一行行的数据
			String[] words = value.split(",");
			for (String word : words) {
				out.collect(word);// 将切割处理的一个个的单词收集起来并返回
			}
		}
	});

	DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String,
	Tuple2<String, Integer>>() {
		@Override
		public Tuple2<String, Integer> map(String value) throws Exception {
			// value就是进来一个个的单词
			return Tuple2.of(value, 1);
		}
	});

	//DataSet中分组是groupBy，DataStream分组是keyBy
	KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(
	new KeySelector<Tuple2<String, Integer>, String>() {
		@Override
		public String getKey(Tuple2<String, Integer> value) throws Exception {
			return value.f0;
		}
	});
	DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

	// 4.输出结果-sink
//        result.print();
	result.addSink(new PrintSinkFunction<>());

	// 5.触发执行-execute
	// env.execute();// DataStream需要调用execute
	env.execute("word count stream");
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

streamNodes = {HashMap@2283}  size = 5
 {Integer@2309} 1 -> {StreamNode@2310} "Source: Collection Source-1"
  key = {Integer@2309} 1
  value = {StreamNode@2310} "Source: Collection Source-1"
 {Integer@2311} 2 -> {StreamNode@2312} "Flat Map-2"
  key = {Integer@2311} 2
  value = {StreamNode@2312} "Flat Map-2"
 {Integer@2313} 3 -> {StreamNode@2314} "Map-3"
  key = {Integer@2313} 3
  value = {StreamNode@2314} "Map-3"
 {Integer@2315} 5 -> {StreamNode@2316} "Keyed Aggregation-5"
  key = {Integer@2315} 5
  value = {StreamNode@2316} "Keyed Aggregation-5"
 {Integer@2317} 6 -> {StreamNode@2318} "Sink: Unnamed-6"
  key = {Integer@2317} 6
  value = {StreamNode@2318} "Sink: Unnamed-6"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

public void go() throws Exception {
	// 1. Create Flink batch execution environment
	ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	env.setParallelism(1);

	// 2. Prepare data source
	DataSet<String> ds = env.fromElements(
			"flink hadoop hive",
			"flink hadoop hive",
			"flink hadoop flink"
	);

	ds.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
		// Split input line into words
		String[] words = value.split(" ");
		for (String word : words) {
			out.collect(new Tuple2<>(word, 1));
		}
	})
	.returns(TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})) 
	.map((MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>) 
		value -> Tuple2.of(value.f0, value.f1))
	.returns(TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}))
	.groupBy(0) // Group by word (Tuple2<String, Integer> is now recognized)
	.sum(1)     // Sum the second field (integer count)
	.print(); 

	env.execute("word count stream");
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

public class OperatorTranslation {

	
	private <T> Operator<T> translate(DataSet<T> dataSet) {
		Operator<T> dataFlowOp;
		if (dataSet instanceof DataSource) {
		
		} else if (dataSet instanceof SingleInputOperator) {
			dataFlowOp = translateSingleInputOperator(singleInputOperator);
		} else if (dataSet instanceof TwoInputOperator) {
			dataFlowOp = translateTwoInputOperator(twoInputOperator);
		} else if (dataSet instanceof BulkIterationResultSet) {
		
		} else if (dataSet instanceof DeltaIterationResultSet) {
		
		} else {
		
		}
		
	}
	
	private <I, O> Operator<O> translateSingleInputOperator(SingleInputOperator<?, ?, ?> op) {
		// 递归调用
		Operator<I> input = translate(typedInput);
		dataFlowOp =
                typedOp.translateToDataFlow(input);
		// ...
		// ...
		return dataFlowOp;
	}
	
	private <I, O> Operator<O> translateTwoInputOperator(SingleInputOperator<?, ?, ?> op) {
		DataSet<I1> typedInput1 = (DataSet<I1>) op.getInput1();
        DataSet<I2> typedInput2 = (DataSet<I2>) op.getInput2();
		// 递归调用
        Operator<I1> input1 = translate(typedInput1);
        Operator<I2> input2 = translate(typedInput2);
		dataFlowOp =
                typedOp.translateToDataFlow(input1, input2);
		// ...
		// ...
		return dataFlowOp;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

                   STREAM 1                             STREAM 2
                   ---------                            ---------
                  +---------+                          +---------+
                  | Source1 |                          | Source2 |
                  +----+----+                          +----+----+
                       |                                      |
                       v                                      v
                  +---------+                           +---------+
                  |  Map1   |                           |  Map2   |
                  +----+----+                           +----+----+
                       |                                      |
                       v                                      v
                  +---------+                           +---------+
                  | Filter1 |                           | Filter2 |
                  +---------+                           +---------+
                       |                                      |
                       |         (Both Chains A & B)          |
                       |______________________________________|
                                      |   (Shuffle/Exchange)
                                      v
                                +------------+
                                |   Join     |
                                | (Stage Break) |
                                +------------+
                                      |
                                      v
                                +------------+
                                | PostJoinMap|
                                +------------+
                                      |
                                      v
                                +------------+
                                |   Sink     |
                                +------------+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

JobGraph:
  [KafkaSource] ---> [HDFSSink]

DefaultExecutionGraphBuilder:
  For each JobVertex:
    +---------------------+         +---------------------+
    | KafkaSource (Job)   |         | HDFSSink (Job)      |
    +---------------------+         +---------------------+
           |                                 |
           v                                 v
    +---------------------+         +---------------------+
    | ExecutionJobVertex  |         | ExecutionJobVertex  |
    | for KafkaSource     |         | for HDFSSink        |
    | (with 4 subtasks)   |         | (with 4 subtasks)   |
    +---------------------+         +---------------------+
           |                                 |
           |--- Create Intermediate Result  |  <--- Wire input/output connections
           |    (data partitions)            |
           v                                 v
    [IntermediateResultPartitions connecting source and sink]
           |                                 |
           v                                 v
    +-------------------------------------------------+
    |          Fully Constructed ExecutionGraph       |
    +-------------------------------------------------+
                   |
                   v
         Scheduler uses this graph

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

                       JobGraph
                        [Map]
                          |
                ---------------------
                |                   |
     ExecutionJobVertex ("Map")
                |
    -----------------------------------------
    |           |           |             |
ExecutionVertex   ExecutionVertex   ExecutionVertex   ExecutionVertex
    #1              #2              #3              #4
     |               |               |               |
 Execution          Execution       Execution       Execution
 (RUNNING)         (RUNNING)       (RUNNING)       (RUNNING)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

JobGraph (Logical Plan)
   +----------------+      +----------------+      +----------------+
   |    Source      | ---> |     Map        | ---> |     Sink       |
   |  parallelism=4 |      |  parallelism=4 |      |  parallelism=4 |
   +----------------+      +----------------+      +----------------+

   Transformation via DefaultExecutionGraphBuilder
                     |
                     V
ExecutionGraph (Runtime Plan)
   +-----------------------+         +-----------------------+         +-----------------------+
   | ExecutionJobVertex    |         | ExecutionJobVertex    |         | ExecutionJobVertex    |
   | for Source (4 tasks)  |         | for Map (4 tasks)     |         | for Sink (4 tasks)    |
   +-----------------------+         +-----------------------+         +-----------------------+
           |                                  |                                     |
           V                                  V                                     V
   +---------+  +---------+ ... +---------+ +---------+  +---------+ ... +---------+ +---------+  +---------+ ... +---------+
   |  Task 1 |  |  Task 2 | ... |  Task 4 | |  Task 1 |  |  Task 2 | ... |  Task 4 | |  Task 1 |  |  Task 2 | ... |  Task 4 |
   +---------+  +---------+ ... +---------+ +---------+  +---------+ ... +---------+ +---------+  +---------+ ... +---------+