9.3.1 基本转换算子
- 映射（map）
map是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素，如图所示。
我们只需要基于DataStrema调用map()方法就可以进行转换处理。方法需要传入的参数是接口MapFunction的实现；返回值类型还是DataStream，不过泛型（流中的元素类型）可能改变。
下面的代码用不同的方式，实现了提取Event中的user字段的功能。
publicclass TransMap { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> stream = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); // 传入匿名类，实现MapFunction stream.map(new MapFunction<Event, String>() { @Override public String map(Event e) throws Exception { return e.user; } }); // 传入MapFunction的实现类 stream.map(new UserExtractor()).print(); env.execute(); } publicstaticclass UserExtractor implements MapFunction<Event, String> { @Override public String map(Event e) throws Exception { return e.user; } } }
上面代码中，MapFunction实现类的泛型类型，与输入数据类型和输出数据的类型有关。在实现MapFunction接口的时候，需要指定两个泛型，分别是输入事件和输出事件的类型，还需要重写一个map()方法，定义从一个输入事件转换为另一个输出事件的具体逻辑。
- 过滤（filter）filter转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为true则元素正常输出，若为false则元素被过滤掉，如图所示。
进行filter转换之后的新数据流的数据类型与原数据流是相同的。filter转换需要传入的参数需要实现FilterFunction接口，而FilterFunction内要实现filter()方法，就相当于一个返回布尔类型的条件表达式。
下面的代码会将数据流中用户Mary的浏览行为过滤出来 。
publicclass TransFilter { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> stream = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); // 传入匿名类实现FilterFunction stream.filter(new FilterFunction<Event>() { @Override public boolean filter(Event e) throws Exception { return e.user.equals("Mary"); } }); // 传入FilterFunction实现类 stream.filter(new UserFilter()).print(); env.execute(); } publicstaticclass UserFilter implements FilterFunction<Event> { @Override public boolean filter(Event e) throws Exception { return e.user.equals("Mary"); } } }
- 扁平映射（flatmap)
flatMap操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生0到多个元素。flatMap可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理，如图5-7所示。我们此前WordCount程序的第一步分词操作，就用到了flatMap。
同map一样，flatMap也可以使用Lambda表达式或者FlatMapFunction接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。
flatMap操作会应用在每一个输入事件上面， FlatMapFunction接口中定义了flatMap方法，用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回0个、1个或多个结果数据。因此flatMap并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。所以flatMap方法也可以实现map方法和filter方法的功能，当返回结果是0个的时候，就相当于对数据进行了过滤，当返回结果是1个的时候，相当于对数据进行了简单的转换操作。
flatMap的使用非常灵活，可以对结果进行任意输出，下面就是一个例子：
publicclass TransFlatmap { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> stream = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); stream.flatMap(new MyFlatMap()).print(); env.execute(); } publicstaticclass MyFlatMap implements FlatMapFunction<Event, String> { @Override public void flatMap(Event value, Collector<String> out) throws Exception { if (value.user.equals("Mary")) { out.collect(value.user); } elseif (value.user.equals("Bob")) { out.collect(value.user); out.collect(value.url); } } } }
9.3.2 聚合算子（Aggregation）
在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的信息。比如之前word count程序中，要对每个词出现的频次进行叠加统计。这种操作，计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并——这就是所谓的“聚合”（Aggregation），也对应着MapReduce中的reduce操作。
- 按键分区（keyBy）
对于Flink而言，DataStream是没有直接进行聚合的API的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在Flink中，要做聚合，需要先进行分区；这个操作就是通过keyBy来完成的。
keyBy是聚合前必须要用到的一个算子。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
keyBy是聚合前必须要用到的一个算子。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
基于不同的key，流中的数据将被分配到不同的分区中去，如图所示；这样一来，所有具有相同的key的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个slot中进行处理了。
在内部，是通过计算key的哈希值（hash code），对分区数进行取模运算来实现的。所以这里key如果是POJO的话，必须要重写hashCode()方法。
keyBy()方法需要传入一个参数，这个参数指定了一个或一组key。有很多不同的方法来指定key：比如对于Tuple数据类型，可以指定字段的位置或者多个位置的组合；对于POJO类型，可以指定字段的名称（String）；另外，还可以传入Lambda表达式或者实现一个键选择器（KeySelector），用于说明从数据中提取key的逻辑。
我们可以以id作为key做一个分区操作，代码实现如下：
publicclass TransKeyBy { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> stream = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); // 使用Lambda表达式 KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user); // 使用匿名类实现KeySelector KeyedStream<Event, String> keyedStream1 = stream.keyBy(new KeySelector<Event, String>() { @Override public String getKey(Event e) throws Exception { return e.user; } }); env.execute(); } }
需要注意的是，keyBy得到的结果将不再是DataStream，而是会将DataStream转换为KeyedStream。KeyedStream可以认为是“分区流”或者“键控流”，它是对DataStream按照key的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定key的类型。
KeyedStream也继承自DataStream，所以基于它的操作也都归属于DataStream API。但它跟之前的转换操作得到的SingleOutputStreamOperator不同，只是一个流的分区操作，并不是一个转换算子。KeyedStream是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如sum，reduce）。
- 简单聚合
有了按键分区的数据流KeyedStream，我们就可以基于它进行聚合操作了。Flink为我们内置实现了一些最基本、最简单的聚合API，主要有以下几种：
sum()：在输入流上，对指定的字段做叠加求和的操作。 min()：在输入流上，对指定的字段求最小值。 max()：在输入流上，对指定的字段求最大值。 minBy()：与min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而minBy()则会返回包含字段最小值的整条数据。 maxBy()：与max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数；但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。
对于元组类型的数据，可以使用这两种方式来指定字段。需要注意的是，元组中字段的名称，是以f0、f1、f2、…来命名的。
如果数据流的类型是POJO类，那么就只能通过字段名称来指定，不能通过位置来指定了。
publicclass TransAggregation { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> stream = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); stream.keyBy(e -> e.user).max("timestamp"); // 指定字段名称 env.execute(); } }
简单聚合算子返回的，同样是一个SingleOutputStreamOperator，也就是从KeyedStream又转换成了常规的DataStream。所以可以这样理解：keyBy和聚合是成对出现的，先分区、后聚合，得到的依然是一个DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不变。
一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个key的数据流上。
- 归约聚合（reduce）
如果说简单聚合是对一些特定统计需求的实现，那么reduce算子就是一个一般化的聚合统计操作了。它可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。
与简单聚合类似，reduce操作也会将KeyedStream转换为DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。
调用KeyedStream的reduce方法时，需要传入一个参数，实现ReduceFunction接口。接口在源码中的定义如下：
publicinterface ReduceFunction<T> extends Function, Serializable { T reduce(T value1, T value2) throws Exception; }
ReduceFunction接口里需要实现reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件；所以，对于一组数据，我们可以先取两个进行合并，然后再将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据，这也就是reduce“归约”的含义。在流处理的底层实现过程中，实际上是将中间“合并的结果”作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。
我们可以单独定义一个函数类实现ReduceFunction接口，也可以直接传入一个匿名类。当然，同样也可以通过传入Lambda表达式实现类似的功能。
与简单聚合类似，reduce操作也会将KeyedStream转换为DataStrema。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。
下面我们来看一个稍复杂的例子。
我们将数据流按照用户id进行分区，然后用一个reduce算子实现sum的功能，统计每个用户访问的频次；进而将所有统计结果分到一组，用另一个reduce算子实现maxBy的功能，记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。
publicclass TransReduce { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 这里的ClickSource()使用了之前自定义数据源小节中的ClickSource() env.addSource(new ClickSource()) // 将Event数据类型转换成元组类型 .map(new MapFunction<Event, Tuple2<String, Long>>() { @Override public Tuple2<String, Long> map(Event e) throws Exception { return Tuple2.of(e.user, 1L); } }) .keyBy(r -> r.f0) // 使用用户名来进行分流 .reduce(new ReduceFunction<Tuple2<String, Long>>() { @Override public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception { // 每到一条数据，用户pv的统计值加1 return Tuple2.of(value1.f0, value1.f1 + value2.f1); } }) .keyBy(r -> true) // 为每一条数据分配同一个key，将聚合结果发送到一条流中去 .reduce(new ReduceFunction<Tuple2<String, Long>>() { @Override public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception { // 将累加器更新为当前最大的pv统计值，然后向下游发送累加器的值 return value1.f1 > value2.f1 ? value1 : value2; } }) .print(); env.execute(); } }
reduce同简单聚合算子一样，也要针对每一个key保存状态。因为状态不会清空，所以我们需要将reduce算子作用在一个有限key的流上。
9.3.3 用户自定义函数（UDF）
在前面的介绍我们可以发现，Flink的DataStream API编程风格其实是一致的：基本上都是基于DataStream调用一个方法，表示要做一个转换操作；方法需要传入一个参数，这个参数都是需要实现一个接口。很容易发现，这些接口有一个共同特点：全部都以算子操作名称 + Function命名，例如源算子需要实现SourceFunction接口，map算子需要实现MapFunction接口，reduce算子需要实现ReduceFunction接口。
这些接口又都继承自Function接口。所以我们不仅可以通过自定义函数类或者匿名类来实现接口，也可以直接传入Lambda表达式。这就是所谓的用户自定义函数（user-defined function，UDF）。
接下来我们就对这几种编程方式做一个梳理总结。
- 函数类（Function Classes）
对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等
所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。之前我们对于API的练习，主要就是基于这种方式。
下面例子实现了FilterFunction接口，用来从用户的点击数据中筛选包含“home”的内容：
publicclass TransFunctionUDF { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStream<String> clicks = env.readTextFile("clicks.csv"); DataStream<String> stream = clicks.filter(new FlinkFilter()); stream.print(); env.execute(); } publicstaticclass FlinkFilter implements FilterFunction<String> { @Override public boolean filter(String value) throws Exception { return value.contains("home"); } }
当然还可以通过匿名类来实现FilterFunction接口：
DataStream<String> stream = clicks.filter(new FilterFunction<String>() { @Override public boolean filter(String value) throws Exception { return value.contains("home"); } });
为了类可以更加通用，我们还可以将用于过滤的关键字"home"抽象出来作为类的属性，调用构造方法时传进去。
DataStream<String> clicks = env.readTextFile("clicks.csv"); DataStream<String> stream = clicks.filter(new KeyWordFilter("home")); publicstaticclass KeyWordFilter implements FilterFunction<String> { private String keyWord; KeyWordFilter(String keyWord) { this.keyWord = keyWord; } @Override public boolean filter(String value) throws Exception { return value.contains(this.keyWord); } }
- 匿名函数（Lambda）
匿名函数（Lambda表达式）是Java 8 引入的新特性，方便我们更加快速清晰地写代码。Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。
Flink 的所有算子都可以使用 Lambda 表达式的方式来进行编码，但是，当 Lambda 表达式使用 Java 的泛型时，我们需要显式的声明类型信息。
下例演示了如何使用Lambda表达式来实现一个简单的 map() 函数，我们使用 Lambda 表达式来计算输入的平方。在这里，我们不需要声明 map() 函数的输入 i 和输出参数的数据类型，因为 Java 编译器会对它们做出类型推断。
publicclass TransFunctionLambda { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<Event> clicks = env.fromElements( new Event("Mary", "./home", 1000L), new Event("Bob", "./cart", 2000L) ); //map函数使用Lambda表达式，不需要进行类型声明 DataStream<String> stream = clicks.map(event -> event.url); stream.print(); env.execute(); } }
由于 OUT 是 Integer 类型而不是泛型，所以 Flink 可以从函数签名 OUT map(IN value) 的实现中自动提取出结果的类型信息。
但是对于像 flatMap() 这样的函数，它的函数签名 void flatMap(IN value, Collectorout) 被 Java 编译器编译成了 void flatMap(IN value, Collector out)，也就是说将Collector的泛型信息擦除掉了。这样 Flink 就无法自动推断输出的类型信息了。
同样地，使用map()时只要涉及到泛型擦除也会有同样的问题。比如返回Flink自定义的元组类型，在WordCount程序中就遇到了类似的问题。
一般来说，可以显式地调用.returns()方法，通过明确指定返回类型来解决这个问题：
DataStream<String> stream = clicks .map(event -> Tuple2.of(event.user, 1L)) .returns(Types.TUPLE(Types.STRING, Types.LONG));
当然，也可以用类或者匿名类的方式来解决。
- 富函数类（Rich Function Classes）
“富函数类”也是DataStream API提供的一个函数类的接口，所有的Flink函数类都有其Rich版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction等。
与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
Rich Function有生命周期的概念。典型的生命周期方法有：
open()方法，是Rich Function的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如map()或者filter()方法被调用之前，open()会首先被调用。所以像文件IO的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在open()方法中完成。。 close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。 需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如RichMapFunction中的map()，在每条数据到来后都会触发一次调用。
来看一个例子：
publicclass RichFunctionExample { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); env .fromElements(1,2,3,4) .map(new RichMapFunction<Integer, Integer>() { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始"); } @Override public Integer map(Integer integer) throws Exception { return integer + 1; } @Override public void close() throws Exception { super.close(); System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束"); } }) .print(); env.execute(); } }
