FLATMAP
DataStream → DataStream
flatMap算子和map算子很类似，不同之处在于针对每一个输入事件flatMap可以生成0个、1个或者多个输出元素。事实上，flatMap转换算子是filter和map的泛化。所以flatMap可以实现map和filter算子的功能。图5-3展示了flatMap如何根据输入事件的颜色来做不同的处理。如果输入事件是白色方框，则直接输出。输入元素是黑框，则复制输入。灰色方框会被过滤掉。
flatMap算子将会应用在每一个输入事件上面。对应的FlatMapFunction定义了flatMap()方法，这个方法返回0个、1个或者多个事件到一个Collector集合中，作为输出结果。
// T: the type of input elements
// O: the type of output elements
FlatMapFunction[T, O]
> flatMap(T, Collector[O]): Unit
实例一：
下面的例子展示了在数据分析教程中经常用到的例子，我们用flatMap来实现。使用_来切割传感器ID，比如sensor_1。
scala version
class IdSplitter extends FlatMapFunction[String, String] {
override def flatMap(id: String, out: Collector[String]) : Unit = {
val arr = id.split("_")
arr.foreach(out.collect)
}
}
匿名函数写法
val splitIds = sensorIds
.flatMap(r => r.split("_"))
java version
public static class IdSplitter implements FlatMapFunction<String, String> {
@Override
public void flatMap(String id, Collector<String> out) {
String[] splits = id.split("_");
for (String split : splits) {
out.collect(split);
}
}
}
匿名函数写法：
DataStream<String> splitIds = sensorIds
.flatMap((FlatMapFunction<String, String>)
(id, out) -> { for (String s: id.split("_")) { out.collect(s);}})
// provide result type because Java cannot infer return type of lambda function
// 提供结果的类型，因为Java无法推断匿名函数的返回值类型
.returns(Types.STRING);
实例二：
函数类
对于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)