代码：
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 <host> --port <port>'");
return;
}
/** 获取{@link StreamExecutionEnvironment}的具体实现的实例 */
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
/** 通过连接给定地址和端口, 获取数据流的数据源 */
DataStream<String> text = env.socketTextStream(hostname, port);
/** 对数据流中的数据进行解析、分组、窗口、以及聚合 */
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("word")
.timeWindow(Time.seconds(5), Time.seconds(1))
.reduce(new ReduceFunction<WordWithCount>() {
@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<String> text = env.socketTextStream(hostname, port);
在StreamExecutionEnviroment中：在指定的host和port上构建了一个接受网络数据的数据源
public DataStreamSource<String> socketTextStream(String hostname, int port) {
return socketTextStream(hostname, port, "\n");
}
public DataStreamSource<String> socketTextStream(String hostname, int port, String delimiter) {
return socketTextStream(hostname, port, delimiter, 0);
}
public DataStreamSource<String> 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<T> extends Function, Serializable {
void run(SourceContext<T> ctx) throws Exception;
void cancel();
@Public
interface SourceContext<T> {
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<String> 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);
}
}
StreamExecutionEnvironment:addSource()方法：
public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName) {
return addSource(function, sourceName, null);
}
public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) {
/** 如果传入的输出数据类型信息为null，则尝试提取输出数据的类型信息 */
if (typeInfo == null) {
if (function instanceof ResultTypeQueryable) {
/** 如果传入的function实现了ResultTypeQueryable接口, 则直接通过接口获取 */
typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
} else {
try {
/** 通过反射机制来提取类型信息 */
typeInfo = TypeExtractor.createTypeInfo(
SourceFunction.class,
function.getClass(), 0, null, null);
} catch (final InvalidTypesException e) {
/** 提取失败, 则返回一个MissingTypeInfo实例 */
typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
}
}
}
/** 根据function是否是ParallelSourceFunction的子类实例来判断是否是一个并行数据源节点 */
boolean isParallel = function instanceof ParallelSourceFunction;
/** 闭包清理, 可减少序列化内容, 以及防止序列化出错 */
clean(function);
StreamSource<OUT, ?> 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<T> outTypeInfo, StreamSource<T, ?> 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<T> transformation) {
super(environment, transformation);
}
public DataStream(StreamExecutionEnvironment environment, StreamTransformation<T> 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<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("word")
.timeWindow(Time.seconds(5), Time.seconds(1))
.reduce(new ReduceFunction<WordWithCount>() {
@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 <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) {
TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes(clean(flatMapper),
getType(), Utils.getCallLocationName(), true);
/** 根据传入的flatMapper这个Function，构建StreamFlatMap这个StreamOperator的具体子类实例 */
return transform("Flat Map", outType, new StreamFlatMap<>(clean(flatMapper)));
}
public <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
/** 读取输入转换的输出类型, 如果是MissingTypeInfo, 则及时抛出异常, 终止操作 */
transformation.getOutputType();
OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(
this.transformation,
operatorName,
operator,
outTypeInfo,
environment.getParallelism());
@SuppressWarnings({ "unchecked", "rawtypes" })
SingleOutputStreamOperator<R> 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<T, Tuple> keyBy(String... fields) {
return keyBy(new Keys.ExpressionKeys<>(fields, getType()));
}
先根据传入的字段名数组，以及数据流的输出数据类型信息，构建出用来描述key的ExpressionKeys的实例，ExpressionKeys有两个属性：
/** key字段的列表, FlatFieldDescriptor 描述了每个key, 在所在类型中的位置以及key自身的数据类信息 */
private List<FlatFieldDescriptor> keyFields;
/** 包含key的数据类型的类型信息, 与构造函数入参中的字段顺序一一对应 */
private TypeInformation<?>[] originalKeyTypes;
在获取key的描述之后，继续调用keyBy的重载方法：
private KeyedStream<T, Tuple> keyBy(Keys<T> keys) {
return new KeyedStream<>(this, clean(KeySelectorUtil.getSelectorForKeys(keys,
getType(), getExecutionConfig())));
}
网友评论