之前的四篇文章对Flink常用的算子进行了详细讲解并附上了大量使用案例：

• Flink算子使用方法及实例演示：map、filter和flatMap
• Flink算子使用方法及实例演示：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表达式为：

val lambda = dataStream.flatMap{
  (value: String, out: Collector[String]) => {
    if (value.size > 10) {
      value.split(" ").foreach(out.collect)
    }
  }
}

然后我们看一下源码中的第二种定义：

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关键字结合，能够在语义上更好地描述数据类型：

val data: DataStream[(String, Long, Double)] = [...]
data.flatMapWith {
  case (symbol, timestamp, price) => // ...
}

使用这种API时，需要添加引用：

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方法：获取运行时上下文。每个并行的算子子任务都有一个运行时上下文，上下文记录了这个算子运行过程中的一些信息，包括算子当前的并行度、算子子任务序号、广播数据、累加器、监控数据。最重要的是，我们可以从上下文里获取状态数据。

我们可以看一下源码中的函数签名：

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)