之前的四篇文章对Flink常用的算子进行了详细讲解并附上了大量使用案例：
- Flink单数据流基本转换：map、filter、flatMap
- Flink基于Key的分组转换：keyBy、reduce和aggregations
- Flink多数据流转换：union和connect
- Flink并行度和数据重分配
总结下来不难发现，使用Flink的算子必须进行自定义，自定义时可以使用Lambda表达式，也可以继承并重写函数类。本文将带大家阅读一些Flink源码，并提供具体的算子使用例子。
函数类
对于map、flatMap、reduce等方法，我们可以实现MapFunction、FlatMapFunction、ReduceFunction等interface接口。这些函数类签名中都有泛型参数，用来定义该函数的输入或输出的数据类型。我们要继承这些类，并重写里面的自定义函数。以flatMap对应的FlatMapFunction为例，它在源码中的定义为：
public interface FlatMapFunction<T, O> extends Function, Serializable {
void flatMap(T value, Collector<O> out) throws Exception;
}这是一个接口类，它继承了Flink的Function函数式接口。函数式接口只有一个抽象函数方法（Single Abstract Method），其目的是为了方便Java 8 Lambda表达式的使用。此外，它还继承了Serializable，以便进行序列化，这是因为这些函数在运行过程中要发送到各个TaskManager上，发送前后要进行序列化和反序列化。需要注意的是，使用这些函数时，一定要保证函数内的所有内容都可以被序列化。如果有一些不能被序列化的内容，或者使用接下来介绍的Rich函数类，或者重写Java的序列化和反序列化方法。
进一步观察FlatMapFunction发现，这个这个函数有两个泛型T和O，T是输入，O是输出，在使用时，要设置好对应的输入和输出数据类型。自定义函数最终归结为重写函数flatMap，函数的两个参数也与输入输出的泛型类型对应，即参数value的是flatMap的输入，数据类型是T，参数out是flatMap的输出，我们需要将类型为O的数据写入out。
我们继承FlatMapFunction，并实现flatMap，只对长度大于limit的字符串切词：
// 使用FlatMapFunction实现过滤逻辑，只对字符串长度大于 limit 的内容进行切词
class WordSplitFlatMap(limit: Int) extends FlatMapFunction[String, String] {
override def flatMap(value: String, out: Collector[String]): Unit = {
// split返回一个Array
// 将Array中的每个元素使用Collector.collect收集起来，起到将列表展平的效果
if (value.size > limit) {
value.split(" ").foreach(out.collect)
}
}
}
val dataStream: DataStream[String] = senv.fromElements("Hello World", "Hello this is Flink")
val function = dataStream.flatMap(new WordSplitFlatMap(10))其中，Collector起着收集输出的作用。
Lambda表达式
当不需要处理非常复杂的业务逻辑时，使用Lambda表达式可能是更好的选择，Lambda表达式能让代码更简洁紧凑。Java 8和Scala都对Lambda表达式支持非常好。
对于flatMap，Flink的Scala源码有三种定义，我们先看一下第一种的定义：
def flatMap[R: TypeInformation](fun: (T, Collector[R]) => Unit): DataStream[R] = {...}flatMap输入是泛型T，输出是泛型R，接收一个名为fun的Lambda表达式，fun形如(T, Collector[R] => {...})。
我们继续以切词为例，Lambda表达式为：
代码语言：javascript复制val lambda = dataStream.flatMap{
(value: String, out: Collector[String]) => {
if (value.size > 10) {
value.split(" ").foreach(out.collect)
}
}
}然后我们看一下源码中的第二种定义：
代码语言：javascript复制def flatMap[R: TypeInformation](fun: T => TraversableOnce[R]): DataStream[R] = {...}与之前的不同，这里的Lambda表达式输入是泛型T，输出是一个TraversableOnce[R],TraversableOnce表示这是一个R组成的列表。与之前使用Collector收集输出不同，这里直接输出一个列表，Flink帮我们将列表做了展平。使用TraversableOnce也导致我们无论如何都要返回一个列表，即使是一个空列表，否则无法匹配函数的定义。总结下来，这种场景的Lambda表达式输入是一个T，无论如何输出都是一个R的列表，即使是一个空列表。
// 只对字符串数量大于15的句子进行处理
val longSentenceWords = dataStream.flatMap {
input => {
if (input.size > 15) {
// 输出是 TraversableOnce 因此返回必须是一个列表
// 这里将Array[String]转成了Seq[String]
input.split(" ").toSeq
} else {
// 为空时必须返回空列表，否则返回值无法与TraversableOnce匹配！
Seq.empty
}
}
}在使用Lambda表达式时，我们应该逐渐学会使用Intellij Idea的类型检查和匹配功能。比如在本例中，如果返回值不是一个TraversableOnce，那么Intellij Idea会将该行标红，告知我们输入或输出的类型不匹配。
此外，还有第三种只针对Scala的Lambda表达式使用方法。Flink为了保持Java和Scala API的一致性，一些Scala独有的特性没有被放入标准的API，而是集成到了一个扩展包中。这种API支持类型匹配的偏函数（Partial Function），结合case关键字结合，能够在语义上更好地描述数据类型：
代码语言：javascript复制val data: DataStream[(String, Long, Double)] = [...]
data.flatMapWith {
case (symbol, timestamp, price) => // ...
}使用这种API时，需要添加引用：
代码语言：javascript复制import org.apache.flink.streaming.api.scala.extensions._这种方式给输入定义了变量名和类型，方便阅读者阅读代码，同时也保留了函数式编程的简洁。Spark的大多数算子默认都支持此功能，对于Spark用户来说，迁移到Flink时需要注意这个区别。此外mapWith、filterWith、keyingBy、reduceWith也都支持这种功能。
使用flatMapWith，之前的切词可以实现为：
val flatMapWith = dataStream.flatMapWith {
case (sentence: String) => {
if (sentence.size > 15) {
sentence.split(" ").toSeq
} else {
Seq.empty
}
}
}Rich函数类
在上面两种算子自定义的基础上，Flink还提供了Rich函数类。从名称上来看，这种函数类在普通的函数类上增加了Rich前缀，比如RichMapFunction、RichFlatMapFunction或RichReduceFunction等等。比起普通的函数类，Rich函数类增加了：
open()方法：Flink在算子调用前会执行这个方法，可以用来进行一些初始化工作。close()方法：Flink在算子最后一次调用结束后执行这个方法，可以用来释放一些资源。getRuntimeContext方法：获取运行时上下文。每个并行的算子子任务都有一个运行时上下文，上下文记录了这个算子运行过程中的一些信息，包括算子当前的并行度、算子子任务序号、广播数据、累加器、监控数据。最重要的是，我们可以从上下文里获取状态数据。
我们可以看一下源码中的函数签名：
代码语言：javascript复制public abstract class RichFlatMapFunction<IN, OUT> extends AbstractRichFunction implements FlatMapFunction<IN, OUT>它既实现了FlatMapFunction接口类，又继承了AbstractRichFunction。其中AbstractRichFunction是一个抽象类，有一个成员变量RuntimeContext，有open、close和getRuntimeContext等方法。
我们尝试使用FlatMapFunction并使用一个累加器。在单机环境下，我们可以用一个for循环做累加统计，但是在分布式计算环境下，计算是分布在多台节点上的，每个节点处理一部分数据，因此单纯循环无法满足计算，累加器是大数据框架帮我们实现的一种机制，允许我们在多节点上进行累加统计。
// 使用RichFlatMapFunction实现
// 添加了累加器 Accumulator
class WordSplitRichFlatMap(limit: Int) extends RichFlatMapFunction[String, String] {
// 创建一个累加器
val numOfLines: IntCounter = new IntCounter(0)
override def open(parameters: Configuration): Unit = {
// 在RuntimeContext中注册累加器
getRuntimeContext.addAccumulator("num-of-lines", this.numOfLines)
}
override def flatMap(value: String, out: Collector[String]): Unit = {
// 运行过程中调用累加器
this.numOfLines.add(1)
if(value.size > limit) {
value.split(" ").foreach(out.collect)
}
}
}
val richFunction = dataStream.flatMap(new WordSplitRichFlatMap(10))
val jobExecuteResult = senv.execute("basic flatMap transformation")
// 执行结束后 获取累加器的结果
val lines: Int = jobExecuteResult.getAccumulatorResult("num-of-lines")
println("num of lines: " lines)
