Flink 1.12.2 源码浅析 : SocketWindowWordCount 代码简析

向右看齐 2022-11-17 13:51 28阅读 0赞

### . ###

*  一 .前言
 *  二 .代码解析
 *   *  2.1. StreamExecutionEnvironment.getExecutionEnvironment
     *  2.2. source加载数据
     *  2.3. Source数据实现
     *   *  2.3.1. SourceFunction 接口.
         *  2.3.2. SocketTextStreamFunction 实现
         *  2.3.3. addSource 实现
     *  2.4. flatMap 函数
     *  2.5. keyBy 函数
     *  2.6. window 函数
     *  2.7. reduce 函数
     *  2.8. windowCounts.print()
     *  2.9. setParallelism(1)
     *  2.10. execute 函数

# 一 .前言 #

本文主要分析官方提供的实例代码SocketWindowWordCount的底层实现.  
主要是为了后续Flink中的StreamGraph做铺垫.

*  代码示例 :

public class SocketWindowWordCount { 
 
 	public static void main(String[] args) throws Exception { 
 
 		// the host and the port to connect to
 		final String hostname;
 		final int port;
 		try { 
 			final ParameterTool params = ParameterTool.fromArgs(args);
 			hostname = params.has("hostname") ? params.get("hostname") : "localhost";
 			port = params.getInt("port");
 		} catch (Exception e) { 
 			System.err.println("No port specified. Please run 'SocketWindowWordCount " +
 				"--hostname <hostname> --port <port>', where hostname (localhost by default) " +
 				"and port is the address of the text server");
 			System.err.println("To start a simple text server, run 'netcat -l <port>' and " +
 				"type the input text into the command line");
 			return;
 		}
 
 		// get the execution environment
 		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
 		// get input data by connecting to the socket
 		DataStream<String> text = env.socketTextStream(hostname, port, "\n");
 
 		// parse the data, group it, window it, and aggregate the counts
 		DataStream<WordWithCount> windowCounts = text
 
 				.flatMap(new FlatMapFunction<String, WordWithCount>() { 
 					@Override
 					public void flatMap(String value, Collector<WordWithCount> out) { 
 						for (String word : value.split("\\s")) { 
 							out.collect(new WordWithCount(word, 1L));
 						}
 					}
 				})
 
 				.keyBy(value -> value.word)
 				.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
 
 				.reduce(new ReduceFunction<WordWithCount>() { 
 					@Override
 					public WordWithCount reduce(WordWithCount a, WordWithCount b) { 
 						return new WordWithCount(a.word, a.count + b.count);
 					}
 				});
 
 		// print the results with a single thread, rather than in parallel
 		windowCounts.print().setParallelism(1);
 
 		env.execute("Socket Window WordCount");
 	}
 
 	// ------------------------------------------------------------------------
 
 	/**
 	 * Data type for words with count.
 	 */
 	public static class WordWithCount { 
 
 		public String word;
 		public long count;
 
 		public WordWithCount() { }
 
 		public WordWithCount(String word, long count) { 
 			this.word = word;
 			this.count = count;
 		}
 
 		@Override
 		public String toString() { 
 			return word + " : " + count;
 		}
 	}
 }

# 二 .代码解析 #

## 2.1. StreamExecutionEnvironment.getExecutionEnvironment ##

> final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

构建StreamExecutionEnvironment 是每个Flink代码的令狐, 接下来我们看一下这里都干了啥.

StreamExecutionEnvironment#getExecutionEnvironment
    		-> StreamExecutionEnvironment#getExecutionEnvironment(Configuration configuration)

*  直接构建StreamExecutionEnvironment

/**
 	 * Creates an execution environment that represents the context in which the
 	 * program is currently executed. If the program is invoked standalone, this
 	 * method returns a local execution environment, as returned by
 	 * { @link #createLocalEnvironment(Configuration)}.
 	 *
 	 * When executed from the command line the given configuration is stacked on top of the
 	 * global configuration which comes from the { @code flink-conf.yaml}, potentially overriding
 	 * duplicated options.
 	 *
 	 * @param configuration The configuration to instantiate the environment with.
 	 * @return The execution environment of the context in which the program is
 	 * executed.
 	 */
 	public static StreamExecutionEnvironment getExecutionEnvironment(Configuration configuration) { 
 		return Utils.resolveFactory(threadLocalContextEnvironmentFactory, contextEnvironmentFactory)
 			.map(factory -> factory.createExecutionEnvironment(configuration))
 			.orElseGet(() -> StreamExecutionEnvironment.createLocalEnvironment(configuration));
 	}

## 2.2. source加载数据 ##

> DataStream text = env.socketTextStream(hostname, port, “\\n”);

接下来是通过socket地址获取数据的操作.我们看一下socketTextStream 干了啥…

StreamExecutionEnvironment#socketTextStream(hostname, port, "\n");
    	-> StreamExecutionEnvironment#socketTextStream(String hostname, int port, String delimiter)
    		-> StreamExecutionEnvironment#socketTextStream(String hostname, int port, String delimiter, long maxRetry) 
    			-> StreamExecutionEnvironment#addSource(new SocketTextStreamFunction(hostname, port, delimiter,maxRetry),"Socket Stream")

在这里我们看到其实是构建了一个 SocketTextStreamFunction ,然后使用StreamExecutionEnvironment\#addSource方法构建的.

## 2.3. Source数据实现 ##

我们以一个source的定义好的方法来说明添加数据源的逻辑.  
SocketTextStreamFunction 是读取socket中的数据, SocketTextStreamFunction实现了SourceFunction 接口 .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

### 2.3.1. SourceFunction 接口. ###

在SourceFunction接口是所有自定义Source的接口类.  
这里面只有一个内置的接口SourceContext 和两个接口

*  执行 : void run(SourceContext ctx) throws Exception;
 *  取消 : void cancel();

### 2.3.2. SocketTextStreamFunction 实现 ###

SocketTextStreamFunction 实现了SourceFunction 接口, 所以就必须实现好他定义好的方法 .

*  静态属性

/** Default delay between successive connection attempts. */
        private static final int DEFAULT_CONNECTION_RETRY_SLEEP = 500;
    
        /** Default connection timeout when connecting to the server socket (infinite). */
        private static final int CONNECTION_TIMEOUT_TIME = 0;
    
        // 主机地址
        private final String hostname;
        // 端口
        private final int port;
        // 分隔符
        private final String delimiter;
        // 最大重试次数
        private final long maxNumRetries;
        // 重试间隔
        private final long delayBetweenRetries;
    
        // 构建的socket
        private transient Socket currentSocket;
    
        // 标识符,是否正在运行
        private volatile boolean isRunning = true;

*  构造方法  
    在构建SocketTextStreamFunction的时候,其实就是值做了赋值操作.

public SocketTextStreamFunction(String hostname, int port, String delimiter, long maxNumRetries, long delayBetweenRetries) { 
    		checkArgument(isValidClientPort(port), "port is out of range");
    		checkArgument(maxNumRetries >= -1, "maxNumRetries must be zero or larger (num retries), or -1 (infinite retries)");
    		checkArgument(delayBetweenRetries >= 0, "delayBetweenRetries must be zero or positive");
    
    		this.hostname = checkNotNull(hostname, "hostname must not be null");
    		this.port = port;
    		this.delimiter = delimiter;
    		this.maxNumRetries = maxNumRetries;
    		this.delayBetweenRetries = delayBetweenRetries;
    	}

*  run 方法  
    run方法其实就是构建socket连接,并且读取数据…

@Override
 public void run(SourceContext<String> ctx) throws Exception { 
 
 //构建buffer 缓存...
 final StringBuilder buffer = new StringBuilder();
 
 // 重试次数
 long attempt = 0;
 
 // 是否运行
 while (isRunning) { 
 
 // 构建socket
 try (Socket socket = new Socket()) { 
 
 
 // 设置当前currentSocket
 currentSocket = socket;
 
 LOG.info("Connecting to server socket " + hostname + ':' + port);
 // 设置连接信息和超时时间
 socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
 
 // 读取socket数据...
 try (BufferedReader reader =
 new BufferedReader(new InputStreamReader(socket.getInputStream()))) { 
 
 char[] cbuf = new char[8192];
 int bytesRead;
 while (isRunning && (bytesRead = reader.read(cbuf)) != -1) { 
 buffer.append(cbuf, 0, bytesRead);
 int delimPos;
 
 // 读取数据...
 while (buffer.length() >= delimiter.length()
 && (delimPos = buffer.indexOf(delimiter)) != -1) { 
 
 // 获取数据记录...
 String record = buffer.substring(0, delimPos);
 // truncate trailing carriage return
 if (delimiter.equals("\n") && record.endsWith("\r")) { 
 record = record.substring(0, record.length() - 1);
 }
 // 处理数据
 ctx.collect(record);
 // 清理掉buffer中的缓存数据...
 buffer.delete(0, delimPos + delimiter.length());
 }
 }
 }
 }
 
 // 如果退出当前循环则重试操作...
 // if we dropped out of this loop due to an EOF, sleep and retry
 if (isRunning) { 
 // 重试次数相关处理.
 attempt++;
 if (maxNumRetries == -1 || attempt < maxNumRetries) { 
 LOG.warn(
 "Lost connection to server socket. Retrying in "
 + delayBetweenRetries
 + " msecs...");
 Thread.sleep(delayBetweenRetries);
 } else { 
 // this should probably be here, but some examples expect simple exists of the
 // stream source
 // throw new EOFException("Reached end of stream and reconnects are not
 // enabled.");
 break;
 }
 }
 }
 
 // 最后操作...
 // collect trailing data
 if (buffer.length() > 0) { 
 ctx.collect(buffer.toString());
 }
 }

*  cancel 取消方法

取消方法就是关闭socket通讯…

@Override
    	public void cancel() { 
    		isRunning = false;
    
    		// we need to close the socket as well, because the Thread.interrupt() function will
    		// not wake the thread in the socketStream.read() method when blocked.
    		Socket theSocket = this.currentSocket;
    		if (theSocket != null) { 
    			IOUtils.closeSocket(theSocket);
    		}
    	}

### 2.3.3. addSource 实现 ###

我们知道添加数据源是通过StreamExecutionEnvironment\#addSource方法实现的…

通过最终通过

StreamExecutionEnvironment\#addSource方法中有一个参数 Boundedness 是枚举类, 代表数据源的类型是有界数据\[BOUNDED\]或者是无界\[CONTINUOUS\_UNBOUNDED\]数据.

/** * Ads a data source with a custom type information thus opening a * {@link DataStream}. Only in very special cases does the user need to * support type information. Otherwise use * {@link #addSource(org.apache.flink.streaming.api.functions.source.SourceFunction)} * * @param function * the user defined function * @param sourceName * Name of the data source * @param <OUT> * type of the returned stream * @param typeInfo * the user defined type information for the stream * @return the data stream constructed */
 	public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) { 
 		return addSource(function, sourceName, typeInfo, Boundedness.CONTINUOUS_UNBOUNDED);
 	}

*  addSource最终的调用方法

private <OUT> DataStreamSource<OUT> addSource(
 			final SourceFunction<OUT> function,
 			final String sourceName,
 			@Nullable final TypeInformation<OUT> typeInfo,
 			final Boundedness boundedness) { 
 		checkNotNull(function);
 		checkNotNull(sourceName);
 		checkNotNull(boundedness);
 
 		TypeInformation<OUT> resolvedTypeInfo = getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);
 
 		boolean isParallel = function instanceof ParallelSourceFunction;
 
 		clean(function);
 
 		final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);
 		return new DataStreamSource<>(this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
 	}

在这里我们看到最终返回了一个DataStreamSource 对象. 他继承了了SingleOutputStreamOperator 类型.

关于datasource的内容就先到这, 后续在继续看DataStreamSource相关的实现.

## 2.4. flatMap 函数 ##

接下来我们看具体的代码操作 : flatMap 函数的实现.  
flatMap的入参是一个FlatMapFunction的具体实现，功能就是将接收到的字符串，按空格切割成不同单词，然后每个单词构建一个WordWithCount实例，然后向下游转发，用于后续的数据统计。然后调用DataStream的flatMap方法，进行数据流的转换.

> a、先根据传入的flatMapper这个Function构建一个StreamOperator的具体子类StreamFlatMap的实例；  
> b、根据a中构建的StreamFlatMap的实例，构建出OneInputTransFormation这个StreamTransformation的子类的实例；  
> c、再构建出DataStream的子类SingleOutputStreamOperator的实例。

DataStream<WordWithCount> windowCounts = text
 				.flatMap(new FlatMapFunction<String, WordWithCount>() { 
 					@Override
 					public void flatMap(String value, Collector<WordWithCount> out) { 
 						for (String word : value.split("\\s")) { 
 							out.collect(new WordWithCount(word, 1L));
 						}
 					}
 				})

这里就是根据空格做的拆分,并返回一个数据的集合.

/** * Applies a FlatMap transformation on a {@link DataStream}. The transformation calls a {@link * FlatMapFunction} for each element of the DataStream. Each FlatMapFunction call can return any * number of elements including none. The user can also extend {@link RichFlatMapFunction} to * gain access to other features provided by the {@link * org.apache.flink.api.common.functions.RichFunction} interface. * * @param flatMapper The FlatMapFunction that is called for each element of the DataStream * @param <R> output type * @return The transformed {@link DataStream}. */
 public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) { 
 
 // outType返回一个PojoType类型的TypeInformation
 // PojoType
 // <
 // org.apache.flink.streaming.examples.socket.SocketWindowWordCount$WordWithCount,
 // fields = [count: Long, word: String]
 // >
 TypeInformation<R> outType =
 TypeExtractor.getFlatMapReturnTypes(
 clean(flatMapper), getType(), Utils.getCallLocationName(), true);
 
 // 调用flatMap 方法
 return flatMap(flatMapper, outType);
 }

*  接下来有调用了DataStream方法, 返回一个R

/** * Applies a FlatMap transformation on a {@link DataStream}. The * transformation calls a {@link FlatMapFunction} for each element of the * DataStream. Each FlatMapFunction call can return any number of elements * including none. The user can also extend {@link RichFlatMapFunction} to * gain access to other features provided by the * {@link org.apache.flink.api.common.functions.RichFunction} interface. * * @param flatMapper * The FlatMapFunction that is called for each element of the * DataStream * @param outputType {@link TypeInformation} for the result type of the function. * * @param <R> * output type * @return The transformed {@link DataStream}. */
 	public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper, TypeInformation<R> outputType) { 
 		return transform("Flat Map", outputType, new StreamFlatMap<>(clean(flatMapper)));
 
 	}

*  在这里我们看到了两个类StreamFlatMap和transform方法, 粗略的看一下…
 *  StreamFlatMap 根据输入的数据做转换执行FlatMapFunctions函数, 不细看了…

public class StreamFlatMap<IN, OUT>
 		extends AbstractUdfStreamOperator<OUT, FlatMapFunction<IN, OUT>>
 		implements OneInputStreamOperator<IN, OUT> { 
 
 	private static final long serialVersionUID = 1L;
 
 	private transient TimestampedCollector<OUT> collector;
 
 	public StreamFlatMap(FlatMapFunction<IN, OUT> flatMapper) { 
 		super(flatMapper);
 		chainingStrategy = ChainingStrategy.ALWAYS;
 	}
 
 	@Override
 	public void open() throws Exception { 
 		super.open();
 		collector = new TimestampedCollector<>(output);
 	}
 
 	@Override
 	public void processElement(StreamRecord<IN> element) throws Exception { 
 		collector.setTimestamp(element);
 		userFunction.flatMap(element.getValue(), collector);
 	}
 }

* transform(“Flat Map”, outputType, new StreamFlatMap<>(clean(flatMapper))); 
 flatMap 会调用transform 方法, 我们看一下,这个干了啥…

DataStream#transform
 	-> DataStream#doTransform 方法
 	 	-> 构造OneInputTransformation 添加到StreamExecutionEnvironment中的 List<Transformation<?>> transformations 集合中

protected <R> SingleOutputStreamOperator<R> doTransform(
 String operatorName,
 TypeInformation<R> outTypeInfo,
 StreamOperatorFactory<R> operatorFactory) { 
 
 // read the output type of the input Transform to coax out errors about MissingTypeInfo
 transformation.getOutputType();
 
 // 构造OneInputTransformation 单输入转换
 OneInputTransformation<T, R> resultTransform =
 new OneInputTransformation<>(
 this.transformation,
 operatorName,
 operatorFactory,
 outTypeInfo,
 environment.getParallelism());
 
 @SuppressWarnings({ "unchecked", "rawtypes"})
 SingleOutputStreamOperator<R> returnStream =
 new SingleOutputStreamOperator(environment, resultTransform);
 
 //添加到StreamExecutionEnvironment中的 List<Transformation<?>> transformations 集合中
 getExecutionEnvironment().addOperator(resultTransform);
 
 return returnStream;
 }

* 核心关注点是构建了一个SingleOutputStreamOperator 然后加入到`StreamExecutionEnvironment中的 List<Transformation<?>> transformations 集合中`

返回的是一个SingleOutputStreamOperator 类型的函数…

*  DataStream\#flatMap的入参数一个FlatMapFunction , 我们看一下这个类其实就是Override了flatMap方法,在flatMap中写业务逻辑.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

## 2.5. keyBy 函数 ##

接下来我看代码执行到了`.keyBy(value -> value.word)`函数 .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

直接看DataStream\#keyBy函数 , 在这里面对数据做了一个转换变成了KeyedStream 对象… 在转换的过程中 transformation属性没有任何变化. 是从DataStream转换为KeyedStream

/** * It creates a new {@link KeyedStream} that uses the provided key for partitioning * its operator states. * * @param key * The KeySelector to be used for extracting the key for partitioning * @return The {@link DataStream} with partitioned state (i.e. KeyedStream) */
 	public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key) { 
 		Preconditions.checkNotNull(key);
 		return new KeyedStream<>(this, clean(key));
 	}

## 2.6. window 函数 ##

接下来我看代码执行到了`.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))`函数 .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]  
直接看window函数, 在这里面其实就是把KeyedStream 有转换为了WindowedStream …

/** * Windows this data stream to a {@code WindowedStream}, which evaluates windows * over a key grouped stream. Elements are put into windows by a {@link WindowAssigner}. The * grouping of elements is done both by key and by window. * * A {@link org.apache.flink.streaming.api.windowing.triggers.Trigger} can be defined to * specify when windows are evaluated. However, {@code WindowAssigners} have a default * {@code Trigger} that is used if a {@code Trigger} is not specified. * * @param assigner The {@code WindowAssigner} that assigns elements to windows. * @return The trigger windows data stream. */
 	@PublicEvolving
 	public <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<? super T, W> assigner) { 
 		return new WindowedStream<>(this, assigner);
 	}

## 2.7. reduce 函数 ##

接下来我看代码执行到了reduce函数.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

WindowedStream\#reduce函数

/** * Applies the given window function to each window. The window function is called for each * evaluation of the window for each key individually. The output of the window function is * interpreted as a regular non-windowed stream. * * Arriving data is incrementally aggregated using the given reducer. * * @param reduceFunction The reduce function that is used for incremental aggregation. * @param function The window function. * @param resultType Type information for the result type of the window function. * @return The data stream that is the result of applying the window function to the window. */
 	public <R> SingleOutputStreamOperator<R> reduce(
 		ReduceFunction<T> reduceFunction,
 		WindowFunction<T, R, K, W> function,
 		TypeInformation<R> resultType) { 
 
 		//clean the closures
 		function = input.getExecutionEnvironment().clean(function);
 		reduceFunction = input.getExecutionEnvironment().clean(reduceFunction);
 
 		final String opName = builder.generateOperatorName(reduceFunction, function);
 
 		OneInputStreamOperator<T, R> operator = builder.reduce(reduceFunction, function);
 		return input.transform(opName, resultType, operator);
 	}

在reduce函数里面,最终构造了一个 OneInputStreamOperator (其实是WindowOperator对象)调用父类的DataStream\#transform函数…  
最终又调用到了 DataStream\#doTransform函数

在这里构建了一个SingleOutputStreamOperator 调用StreamExecutionEnvironment\#addOperator方法, 更新StreamExecutionEnvironment中的transformations 属性.

截止到这里 StreamExecutionEnvironment中的transformations 属性.有了两条属性 :

transformations = { ArrayList@3355} size = 2
 
 0 = { OneInputTransformation@3472} "OneInputTransformation{id=2, name='Flat Map', outputType=PojoType<org.apache.flink.streaming.examples.socket.SocketWindowWordCount$WordWithCount, fields = [count: Long, word: String]>, parallelism=1}"
 
 1 = { OneInputTransformation@3613} "OneInputTransformation{id=4, name='Window(TumblingProcessingTimeWindows(5000), ProcessingTimeTrigger, ReduceFunction$1, PassThroughWindowFunction)', outputType=PojoType<org.apache.flink.streaming.examples.socket.SocketWindowWordCount$WordWithCount, fields = [count: Long, word: String]>, parallelism=1}"

然后再调用KeyedStream\#doTransform方法返回SingleOutputStreamOperator操作.

// ------------------------------------------------------------------------
 	// basic transformations
 	// ------------------------------------------------------------------------
 
 	@Override
 	protected <R> SingleOutputStreamOperator<R> doTransform(
 			final String operatorName,
 			final TypeInformation<R> outTypeInfo,
 			final StreamOperatorFactory<R> operatorFactory) { 
 
 		SingleOutputStreamOperator<R> returnStream = super.doTransform(operatorName, outTypeInfo, operatorFactory);
 
 		// inject the key selector and key type
 		OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation();
 		transform.setStateKeySelector(keySelector);
 		transform.setStateKeyType(keyType);
 
 		return returnStream;
 	}

## 2.8. windowCounts.print() ##

print方法构建了一个PrintSinkFunction. 通过DataStream的addSink方法返回一个DataStreamSink.

/** * Writes a DataStream to the standard output stream (stdout). * * For each element of the DataStream the result of {@link Object#toString()} is written. * * NOTE: This will print to stdout on the machine where the code is executed, i.e. the Flink * worker. * * @return The closed DataStream. */
 	@PublicEvolving
 	public DataStreamSink<T> print() { 
 		PrintSinkFunction<T> printFunction = new PrintSinkFunction<>();
 		return addSink(printFunction).name("Print to Std. Out");
 	}

在DataStream的addSink方法中构造了一个StreamSink 对象 , 并更新了 StreamExecutionEnvironment中的transformations 属性 .

/** * Adds the given sink to this DataStream. Only streams with sinks added * will be executed once the {@link StreamExecutionEnvironment#execute()} * method is called. * * @param sinkFunction * The object containing the sink's invoke function. * @return The closed DataStream. */
 	public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) { 
 
 		// read the output type of the input Transform to coax out errors about MissingTypeInfo
 		transformation.getOutputType();
 
 		// configure the type if needed
 		if (sinkFunction instanceof InputTypeConfigurable) { 
 			((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig());
 		}
 
 		StreamSink<T> sinkOperator = new StreamSink<>(clean(sinkFunction));
 
 		DataStreamSink<T> sink = new DataStreamSink<>(this, sinkOperator);
 
 		getExecutionEnvironment().addOperator(sink.getTransformation());
 		return sink;
 	}

截止到这里 StreamExecutionEnvironment中的transformations 属性.有了三条属性 :

transformations = { ArrayList@3355} size = 3
 
 0 = { OneInputTransformation@3472} 
 "OneInputTransformation{ 
 id=2, 
 name='Flat Map', 
 outputType=PojoType<org.apache.flink.streaming.examples.socket.SocketWindowWordCount$WordWithCount, 
 fields = [count: Long, word: String]>, parallelism=1
 }"
 
 1 = { OneInputTransformation@3613} 
 "OneInputTransformation{ 
 id=4, 
 name='Window(TumblingProcessingTimeWindows(5000), ProcessingTimeTrigger, ReduceFunction$1, PassThroughWindowFunction)', 
 outputType=PojoType<org.apache.flink.streaming.examples.socket.SocketWindowWordCount$WordWithCount, fields = [count: Long, word: String]>, 
 parallelism=1
 }"
 
 2 = { LegacySinkTransformation@3682} 
 "LegacySinkTransformation{ 
 id=5, 
 name='Print to Std. Out', 
 outputType=GenericType<java.lang.Object>, 
 parallelism=1
 }"

## 2.9. setParallelism(1) ##

设置并行度.

直接调用了DataStreamSink\#setParallelism方法更新了并行度 parallelism 属性…

/** * Sets the parallelism for this sink. The degree must be higher than zero. * * @param parallelism The parallelism for this sink. * @return The sink with set parallelism. */
 	public DataStreamSink<T> setParallelism(int parallelism) { 
 		transformation.setParallelism(parallelism);
 		return this;
 	}

## 2.10. execute 函数 ##

截止到这里开始出发执行操作… 开启构建StreamGraph 等等一系列操作…

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221022/810ac16e918b4748aef0bf67d8384c01.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221022/e055f415d95043f28c39d7004f2d062b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221022/c8d32610ee2742eb88612979ccbc4225.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221022/3683ab57fb49430ab7fdb5c486c100ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: /images/20221022/ec8cf1949f664c089a8b72dedaaacef9.png