【零基础学flink】flink源码解析【零基础学flink】flink源码解析

本文出处：https://www.jianshu.com/p/628f26cc1458

代码：

public class SocketWindowWordCount { public static void main(String[] args) throws Exception { /** 需要连接的主机名和端口 */ final String hostname; final int port; try { final ParameterTool params = ParameterTool.fromArgs(args); hostname = params.get("hostname"); port = params.getInt("port"); } catch (Exception e) { e.printStackTrace(); System.err.println("Please run 'SocketWindowWordCount --host--port '"); return; }/** 获取{@link StreamExecutionEnvironment}的具体实现的实例 */ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); /** 通过连接给定地址和端口, 获取数据流的数据源 */ DataStream text = env.socketTextStream(hostname, port); /** 对数据流中的数据进行解析、分组、窗口、以及聚合 */ DataStream windowCounts = text .flatMap(new FlatMapFunction() { @Override public void flatMap(String value, Collector out) { for (String word : value.split("\\s")) { out.collect(new WordWithCount(word, 1L)); } } }) .keyBy("word") .timeWindow(Time.seconds(5), Time.seconds(1)) .reduce(new ReduceFunction() { @Override public WordWithCount reduce(WordWithCount a, WordWithCount b) { return new WordWithCount(a.word, a.count + b.count); } }); /** 打印出分析结果 */ windowCounts.print(); /** 懒加载，执行处理程序 */ env.execute("Socket Window WordCount"); }/** 单词和统计次数的数据结构 */ public static class WordWithCount { public String word; public long count; public WordWithCount(String word, long count) { this.word = word; this.count = count; }@Override public String toString() { return word + " : " + count; } } }

数据源
数据源的构建是通过StreamExecutionEnviroment的具体实现的实例来构建的。

DataStream text = env.socketTextStream(hostname, port);

在StreamExecutionEnviroment中：在指定的host和port上构建了一个接受网络数据的数据源

public DataStreamSource socketTextStream(String hostname, int port) { return socketTextStream(hostname, port, "\n"); }public DataStreamSource socketTextStream(String hostname, int port, String delimiter) { return socketTextStream(hostname, port, delimiter, 0); }public DataStreamSource socketTextStream(String hostname, int port, String delimiter, long maxRetry) { return addSource(new SocketTextStreamFunction(hostname, port, delimiter, maxRetry), "Socket Stream"); }

可以看到会根据传入的hostname、port，以及默认的行分隔符”\n”，和最大尝试次数0，构造一个SocketTextStreamFunction实例，并采用默认的数据源节点名称为”Socket Stream”。
SocketTextStreamFunction的类继承图如下所示，可以看出其是SourceFunction的一个子类，而SourceFunction是Flink中数据源的基础接口。

文章图片
也就是：SocketTextStreamFunction 实现了SourceFunction接口，而SourceFunction继承了Function和Serializable两个接口，其中Function也继承了Serializable接口
SourceFunction内部方法：

文章图片
SourceFunction是flink中所有data source必须实现的最基本的接口，可以看到该接口只有run和cancel两个方法，以及一个内部类SourceContext。run方法一般是一个死循环，但是可以通过cancel方法使run方法终止执行（大部分是通过cancel方法改变一个volatile类型的bool变量实现）

@Public public interface SourceFunction extends Function, Serializable { void run(SourceContext ctx) throws Exception; void cancel(); @Public interface SourceContext { void collect(T element); @PublicEvolving void collectWithTimestamp(T element, long timestamp); @PublicEvolving void emitWatermark(Watermark mark); @PublicEvolving void markAsTemporarilyIdle(); Object getCheckpointLock(); void close(); } }

run(SourceContex)方法：就是实现数据获取逻辑的地方，并可以通过传入的参数ctx（ctx是SourceContext类型）实现向下游节点的数据转发。
cancel()方法：则是用来取消数据源的数据产生，一般在run方法中，会存在一个循环来持续产生数据，而cancel方法则可以使得该循环终止。
具体而言，我们可以研究下SocketTextStreamFunction的具体实现（也就是主要看其run方法的具体实现）：
先看下类的介绍：

/** * A source function that reads strings from a socket. The source will read bytes from the socket * stream and convert them to characters, each byte individually. When the delimiter character is * received, the function will output the current string, and begin a new string. */

SocketTextStreamFuction主要是从socket读取byte数据，读取到的byte数据会被转换为字符，在接收到分隔符前，读取到的字符会被认为一个String；接收到分隔符后，也就意味着一个新的string即将到来。
下面是SocketTextStreamFunction中的几个主要成员属性：

/** 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; private transient Socket currentSocket; private volatile boolean isRunning = true;

isRunning 就是上面提到的那个volatile修饰的bool标志，delimiter由构造器传入，即两个String使用什么分隔的。下面是所有data source类的核心，即run方法的实现

public void run(SourceContext ctx) throws Exception { final StringBuilder buffer = new StringBuilder(); long attempt = 0; //重试次数 /** 这里是第一层循环，只要当前处于运行状态，该循环就不会退出，会一直循环 */ while (isRunning) { try (Socket socket = new Socket()) { /** 对指定的hostname和port，建立Socket连接，并构建一个BufferedReader，用来从Socket中读取数据 */ currentSocket = socket; LOG.info("Connecting to server socket " + hostname + ':' + port); socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME); BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); char[] cbuf = new char[8192]; int bytesRead; /** 这里是第二层循环，对运行状态进行了双重校验，同时对从Socket中读取的字节数进行判断 */ while (isRunning && (bytesRead = reader.read(cbuf)) != -1) { buffer.append(cbuf, 0, bytesRead); int delimPos; /** 这里是第三层循环，就是对从Socket中读取到的数据，按行分隔符进行分割，并将每行数据作为一个整体字符串向下游转发 */ while (buffer.length() >= delimiter.length() && (delimPos = buffer.indexOf(delimiter)) != -1) { String record = buffer.substring(0, delimPos); if (delimiter.equals("\n") && record.endsWith("\r")) { record = record.substring(0, record.length() - 1); } /** 用入参ctx，进行数据的转发 */ ctx.collect(record); buffer.delete(0, delimPos + delimiter.length()); } } } /** 如果由于遇到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 { break; } } } /** 在最外层的循环都退出后，最后检查下缓存中是否还有数据，如果有，则向下游转发 */ if (buffer.length() > 0) { ctx.collect(buffer.toString()); } }

cancel方法：

public void cancel() { isRunning = false; Socket theSocket = this.currentSocket; /** 如果当前socket不为null，则进行关闭操作 */ if (theSocket != null) { IOUtils.closeSocket(theSocket); } }

【【零基础学flink】flink源码解析】StreamExecutionEnvironment:addSource()方法：

public DataStreamSource addSource(SourceFunction function, String sourceName) { return addSource(function, sourceName, null); }public DataStreamSource addSource(SourceFunction function, String sourceName, TypeInformation typeInfo) { /** 如果传入的输出数据类型信息为null，则尝试提取输出数据的类型信息 */ if (typeInfo == null) { if (function instanceof ResultTypeQueryable) { /** 如果传入的function实现了ResultTypeQueryable接口, 则直接通过接口获取 */ typeInfo = ((ResultTypeQueryable) function).getProducedType(); } else { try { /** 通过反射机制来提取类型信息 */ typeInfo = TypeExtractor.createTypeInfo( SourceFunction.class, function.getClass(), 0, null, null); } catch (final InvalidTypesException e) { /** 提取失败, 则返回一个MissingTypeInfo实例 */ typeInfo = (TypeInformation) new MissingTypeInfo(sourceName, e); } } } /** 根据function是否是ParallelSourceFunction的子类实例来判断是否是一个并行数据源节点 */ boolean isParallel = function instanceof ParallelSourceFunction; /** 闭包清理, 可减少序列化内容, 以及防止序列化出错 */ clean(function); StreamSource sourceOperator; /** 根据function是否是StoppableFunction的子类实例, 来决定构建不同的StreamOperator */ if (function instanceof StoppableFunction) { sourceOperator = new StoppableStreamSource<>(cast2StoppableSourceFunction(function)); } else { sourceOperator = new StreamSource<>(function); } /** 返回一个新构建的DataStreamSource实例 */ return new DataStreamSource<>(this, typeInfo, sourceOperator, isParallel, sourceName); }

通过对addSource重载方法的依次调用，最后得到了一个DataStreamSource的实例。
TypeInformation是Flink的类型系统中的核心类，用作函数输入和输出的类型都需要通过TypeInformation来表示，TypeInformation可以看做是数据类型的一个工具，可以通过它获取对应数据类型的序列化器和比较器等。
StreamSource的类继承图如下所示：

文章图片
image 上图可以看出StreamSource是StreamOperator接口的一个具体实现类，其构造函数的入参就是一个SourceFunction的子类实例，这里就是前面介绍过的SocketTextStreamFunciton的实例，构造过程如下：

public StreamSource(SRC sourceFunction) { super(sourceFunction); this.chainingStrategy = ChainingStrategy.HEAD; }public AbstractUdfStreamOperator(F userFunction) { this.userFunction = requireNonNull(userFunction); checkUdfCheckpointingPreconditions(); }private void checkUdfCheckpointingPreconditions() { if (userFunction instanceof CheckpointedFunction && userFunction instanceof ListCheckpointed) { throw new IllegalStateException("User functions are not allowed to implement AND ListCheckpointed."); } }

把传入的userFunction赋值给自己的属性变量，并对传入的userFunction做了校验工作，然后将链接策略设置为HEAD。
Flink中为了优化执行效率，会对数据处理链中的相邻节点会进行合并处理，链接策略有三种：
ALWAYS —— 尽可能的与前后节点进行链接；
NEVER —— 不与前后节点进行链接；
HEAD —— 只能与后面的节点链接，不能与前面的节点链接。
作为数据源的源头，是最顶端的节点了，所以只能采用HEAD或者NEVER，对于StreamSource，采用的是HEAD策略。
StreamOperator是Flink中流操作符的基础接口，其抽象子类AbstractStreamOperator实现了一些公共方法，用户自定义的数据处理逻辑会被封装在StreamOperator的具体实现子类中。
在sourceOperator变量被赋值后，即开始进行DataStreamSource的实例构建，并作为数据源构造调用的返回结果。

return new DataStreamSource<>(this, typeInfo, sourceOperator, isParallel, sourceName);

DataStreamSource的类继承图如下所示，是具有一个预定义输出类型的DataStream。

文章图片
image 在Flink中，DataStream描述了一个具有相同数据类型的数据流，其提供了数据操作的各种API，如map、reduce等，通过这些API，可以对数据流中的数据进行各种操作，DataStreamSource的构建过程如下：

public DataStreamSource(StreamExecutionEnvironment environment, TypeInformation outTypeInfo, StreamSource operator, boolean isParallel, String sourceName) { super(environment, new SourceTransformation<>(sourceName, operator, outTypeInfo, environment.getParallelism())); this.isParallel = isParallel; if (!isParallel) { setParallelism(1); } }protected SingleOutputStreamOperator(StreamExecutionEnvironment environment, StreamTransformation transformation) { super(environment, transformation); }public DataStream(StreamExecutionEnvironment environment, StreamTransformation transformation) { this.environment = Preconditions.checkNotNull(environment, "Execution Environment must not be null."); this.transformation = Preconditions.checkNotNull(transformation, "Stream Transformation must not be null."); }

可见构建过程就是初始化了DataStream中的environment和transformation这两个属性。
其中transformation赋值的是SourceTranformation的一个实例，SourceTransformation是StreamTransformation的子类，而StreamTransformation则描述了创建一个DataStream的操作。对于每个DataStream，其底层都是有一个StreamTransformation的具体实例的，所以在DataStream在构造初始时会为其属性transformation设置一个具体的实例。并且DataStream的很多接口的调用都是直接调用的StreamTransformation的相应接口，如并行度、id、输出数据类型信息、资源描述等。
通过上述过程，根据指定的hostname和port进行数据产生的数据源就构造完成了，获得的是一个DataStreamSource的实例，描述的是一个输出数据类型是String的数据流的源。
在上述的数据源的构建过程中，出现Function(SourceFunction)、StreamOperator、StreamTransformation、DataStream这四个接口：
Function接口：用户通过继承该接口的不同子类来实现用户自己的数据处理逻辑，如上述中实现了SourceFunction这个子类，来实现从指定hostname和port来接收数据，并转发字符串的逻辑；
StreamOperator接口：数据流操作符的基础接口，该接口的具体实现子类中，会有保存用户自定义数据处理逻辑的函数的属性，负责对userFunction的调用，以及调用时传入所需参数，比如在StreamSource这个类中，在调用SourceFunction的run方法时，会构建一个SourceContext的具体实例，作为入参，用于run方法中，进行数据的转发；
StreamTransformation接口：该接口描述了构建一个DataStream的操作，以及该操作的并行度、输出数据类型等信息，并有一个属性，用来持有StreamOperator的一个具体实例；
DataStream：描述的是一个具有相同数据类型的数据流，底层是通过具体的StreamTransformation来实现，其负责提供各种对流上的数据进行操作转换的API接口。
通过上述的关系，最终用户自定义数据处理逻辑的函数，以及并行度、输出数据类型等就都包含在了DataStream中，而DataStream也就可以很好的描述一个具体的数据流了。
上述四个接口的包含关系是这样的：Function –> StreamOperator –> StreamTransformation –> DataStream。
通过数据源的构造，理清Flink数据流中的几个接口的关系后，接下来再来看如何在数据源上进行各种操作，达到最终的数据统计分析的目的。
操作数据流
进行具体的转换操作:

DataStream windowCounts = text .flatMap(new FlatMapFunction() { @Override public void flatMap(String value, Collector out) { for (String word : value.split("\\s")) { out.collect(new WordWithCount(word, 1L)); } } }) .keyBy("word") .timeWindow(Time.seconds(5), Time.seconds(1)) .reduce(new ReduceFunction() { @Override public WordWithCount reduce(WordWithCount a, WordWithCount b) { return new WordWithCount(a.word, a.count + b.count); } });

对数据流做了四次操作，分别是flatMap、keyBy、timeWindow、reduce。
flatMap转换 flatMap的入参是一个FlatMapFunction的具体实现，功能就是将接收到的字符串，按空格切割成不同单词，然后每个单词构建一个WordWithCount实例，然后向下游转发，用于后续的数据统计。然后调用DataStream的flatMap方法，进行数据流的转换，如下：

public SingleOutputStreamOperator flatMap(FlatMapFunction flatMapper) { TypeInformation outType = TypeExtractor.getFlatMapReturnTypes(clean(flatMapper), getType(), Utils.getCallLocationName(), true); /** 根据传入的flatMapper这个Function，构建StreamFlatMap这个StreamOperator的具体子类实例 */ return transform("Flat Map", outType, new StreamFlatMap<>(clean(flatMapper))); }public SingleOutputStreamOperator transform(String operatorName, TypeInformation outTypeInfo, OneInputStreamOperator operator) { /** 读取输入转换的输出类型, 如果是MissingTypeInfo, 则及时抛出异常, 终止操作 */ transformation.getOutputType(); OneInputTransformation resultTransform = new OneInputTransformation<>( this.transformation, operatorName, operator, outTypeInfo, environment.getParallelism()); @SuppressWarnings({ "unchecked", "rawtypes" }) SingleOutputStreamOperator returnStream = new SingleOutputStreamOperator(environment, resultTransform); getExecutionEnvironment().addOperator(resultTransform); return returnStream; }

除了构建出了SingleOutputStreamOperator这个实例为并返回外，还有代码：

getExecutionEnvironment().addOperator(resultTransform); public void addOperator(StreamTransformation transformation) { Preconditions.checkNotNull(transformation, "transformation must not be null."); this.transformations.add(transformation); }

就是将上述构建的OneInputTransFormation的实例，添加到了StreamExecutionEnvironment的属性transformations这个类型为List。
keyBy 这里的keyBy转换，入参是一个字符串”word”，意思是按照WordWithCount中的word字段进行分区操作。

public KeyedStream keyBy(String... fields) { return keyBy(new Keys.ExpressionKeys<>(fields, getType())); }

先根据传入的字段名数组，以及数据流的输出数据类型信息，构建出用来描述key的ExpressionKeys的实例，ExpressionKeys有两个属性：

/** key字段的列表, FlatFieldDescriptor 描述了每个key, 在所在类型中的位置以及key自身的数据类信息 */ private List keyFields; /** 包含key的数据类型的类型信息, 与构造函数入参中的字段顺序一一对应 */ private TypeInformation[] originalKeyTypes;

在获取key的描述之后，继续调用keyBy的重载方法：

private KeyedStream keyBy(Keys keys) { return new KeyedStream<>(this, clean(KeySelectorUtil.getSelectorForKeys(keys, getType(), getExecutionConfig()))); }