DataStream（二）

5.3.2 聚合算子（Aggregation）

5.3.3 用户自定义函数（UDF）

3. 扁平映射（flatMap）

flatMap 操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap 可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。我们此前 WordCount 程序的第一步分词操作，就用到了flatMap。

同 map 一样，flatMap 也可以使用 Lambda 表达式或者 FlatMapFunction 接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。

flatMap 操作会应用在每一个输入事件上面， FlatMapFunction 接口中定义了 flatMap 方法，用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回 0 个、1 个或多个结果数据。因此 flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。

所以 flatMap 方法也可以实现 map 方法和 filter 方法的功能，当返回结果是 0 个的时候，就相当于对数据进行了过滤，当返回结果是 1 个的时候，相当于对数据进行了简单的转换操作。

代码实现：

package com.atguigu.chapter05;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class TransFromFlatMapTest {//1.实现自定义的FlatMapFunction接口public static class MyFlatMap implements FlatMapFunction<Event,String> {@Overridepublic void flatMap(Event event, Collector<String> collector) throws Exception {collector.collect(event.user);collector.collect(event.url);collector.collect(event.timestamp.toString());}}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),new Event("Alice","./prod?id=100",3000L));/*SingleOutputStreamOperator<String> result = stream.flatMap(new MyFlatMap());result.print();*///2.直接传入Lambda表达式(这里注意，要显示返回数据类型，否则报错)stream.flatMap((Event event,Collector<String> out)-> {out.collect(event.user);out.collect(event.url);out.collect(event.timestamp.toString());}).returns(new TypeHint<String>() {}).print();env.execute();}
}

5.3.2 聚合算子（Aggregation）

直观上看，基本转换算子确实是在“转换”——因为它们都是基于当前数据，去做了处理和输出。而在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的信息。比如之前 word count 程序中，要对每个词出现的频次进行叠加统计。这种操作，计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并 ——这就是所谓的“聚合”（Aggregation），也对应着 MapReduce 中的 reduce 操作。

1. 按键分区（keyBy）

对于 Flink 而言，DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区；这个操作就是通过 keyBy 来完成的。

keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。

基于不同的 key，流中的数据将被分配到不同的分区中去，如图所示；这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot中进行处理了。

在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。所以这里 key 如果是 POJO 的话，必须要重写 hashCode()方法。 keyBy()方法需要传入一个参数，这个参数指定了一个或一组 key。有很多不同的方法来指定 key：比如对于 Tuple 数据类型，可以指定字段的位置或者多个位置的组合；对于 POJO 类型，可以指定字段的名称（String）；另外，还可以传入 Lambda 表达式或者实现一个键选择器（KeySelector），用于说明从数据中提取 key 的逻辑。

import org.apache.flink.api.java.functions.KeySelector; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.datastream.KeyedStream; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class TransKeyByTest { 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）；而且它可以将当前算子任务的状态（state）也按照 key 进行划分、限定为仅对当前 key 有效。

2. 简单聚合

有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink 为我们内置实现了一些最基本、最简单的聚合 API，主要有以下几种：

⚫ sum()：在输入流上，对指定的字段做叠加求和的操作。

⚫ min()：在输入流上，对指定的字段求最小值。

⚫ max()：在输入流上，对指定的字段求最大值。

⚫ minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。

⚫ maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。

简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数；但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。

对于元组类型的数据，同样也可以使用这两种方式来指定字段。需要注意的是，元组中字段的名称，是以 f0、f1、f2、…来命名的。

package com.atguigu.chapter05;import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;//单聚合测试
public class TransFromSimpleAggTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Event> stream=env.fromElements(new Event("Bob","./cart",2000L),new Event("Alice","./prod?id=100",3000L),new Event("Bob","./prod?id=1",3300L),new Event("Bob","./home",3500L),new Event("Bob","./prod?id=3",4200L),new Event("Bob","./prod?id=2",3800L));/*//按键分组之后进行聚合，提取当前用户最近一次访问数据stream.keyBy(new KeySelector<Event,String>() {@Overridepublic String getKey(Event event) throws Exception {return event.user;}}).max("timestamp").print("max: ");*///Lambda表达式//maxBy：返回包含字段最大值的整条数据stream.keyBy(data -> data.user).maxBy("timestamp").print("maxBy: ");env.execute();}
}

简单聚合算子返回的，同样是一个 SingleOutputStreamOperator，也就是从 KeyedStream 又转换成了常规的 DataStream。所以可以这样理解：keyBy 和聚合是成对出现的，先分区、后聚合，得到的依然是一个 DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不变。

一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个 key 的数据流上。

3. 归约聚合（reduce）

educe 算子就是一个一般化的聚合统计操作了。从大名鼎鼎的 MapReduce 开始，我们对 reduce 操作就不陌生：它可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。

与简单聚合类似，reduce 操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。

package com.atguigu.chapter05;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class TransFromReduceTest {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Event> stream=env.fromElements(new Event("Bob","./cart",2000L),new Event("Alice","./prod?id=100",3000L),new Event("Bob","./prod?id=1",3300L),new Event("Bob","./home",3500L),new Event("Bob","./prod?id=3",4200L),new Event("Bob","./prod?id=2",3800L));//1.统计每个用户的访问频次SingleOutputStreamOperator<Tuple2<String, Long>> clicksByUser = stream.map(new MapFunction<Event, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(Event event) throws Exception {return Tuple2.of(event.user, 1L);}}).keyBy(data -> data.f0).reduce(new ReduceFunction<Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {return Tuple2.of(value1.f0, value1.f1 + value2.f1);}});//2.选取当前最活跃的用户SingleOutputStreamOperator<Tuple2<String, Long>> result = clicksByUser.keyBy(data -> "key").reduce(new ReduceFunction<Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {return value1.f1 > value2.f1 ? value1 : value2;}});result.print();env.execute();}
}

reduce 同简单聚合算子一样，也要针对每一个 key 保存状态。因为状态不会清空，所以我们需要将 reduce 算子作用在一个有限 key 的流上。

5.3.3 用户自定义函数（UDF）

Flink 的 DataStream API 编程风格其实是一致的：基本上都是基于 DataStream 调用一个方法，表示要做一个转换操作；方法需要传入一个参数，这个参数都是需要实现一个接口。

很容易发现，这些接口有一个共同特点：全部都以算子操作名称 + Function 命名。

我们不仅可以通过自定义函数类或者匿名类来实现接口，也可以直接传入 Lambda 表达式。这就是所谓的用户自定义函数（user-defined function，UDF）。

1. 函数类（Function Classes）

对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类，例如 MapFunction、FilterFunction、ReduceFunction 等。

所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。

下面例子实现了 FilterFunction 接口，用来筛选 url 中包含“home”的事件：

代码实现：

import org.apache.flink.api.common.functions.FilterFunction; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class TransFunctionUDFTest { 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) ); DataStream<Event> stream = clicks.filter(new FlinkFilter()); stream.print(); env.execute(); } public static class FlinkFilter implements FilterFunction<Event> { @Override public boolean filter(Event value) throws Exception { return value.url.contains("home"); } } 
}

当然还可以通过匿名类来实现 FilterFunction 接口：

DataStream<String> stream = clicks.filter(new FilterFunction<Event>() { @Override public boolean filter(Event value) throws Exception { return value.url.contains("home"); } 
});

为了类可以更加通用，我们还可以将用于过滤的关键字"home"抽象出来作为类的属性，调用构造方法时传进去。

DataStream<Event> stream = clicks.filter(new KeyWordFilter("home")); public static class KeyWordFilter implements FilterFunction<Event> { private String keyWord; KeyWordFilter(String keyWord) { this.keyWord = keyWord; } @Override public boolean filter(Event value) throws Exception { return value.url.contains(this.keyWord); } 
}

2. 匿名函数（Lambda）

匿名函数（Lambda 表达式）是 Java 8 引入的新特性，方便我们更加快速清晰地写代码。 Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。 Flink 的所有算子都可以使用 Lambda 表达式的方式来进行编码，但是，当 Lambda 表达式使用 Java 的泛型时，我们需要显式的声明类型信息。

下例演示了如何使用 Lambda 表达式来实现一个简单的 map() 函数，我们使用 Lambda表达式来计算输入的平方。在这里，我们不需要声明 map() 函数的输入 i 和输出参数的数据类型，因为 Java 编译器会对它们做出类型推断。

import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.java.tuple.Tuple2; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class TransFunctionLambdaTest { 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> stream1 = clicks.map(event -> event.url); stream1.print(); env.execute(); } 
}

由于 OUT 是 String 类型而不是泛型，所以 Flink 可以从函数签名 OUT map(IN value)

的实现中自动提取出结果的类型信息。

但是对于像 flatMap() 这样的函数，它的函数签名 void flatMap(IN value, Collector out) 被 Java 编译器编译成了 void flatMap(IN value, Collector out)，也就是说将 Collector 的泛型信息擦除掉了。这样 Flink 就无法自动推断输出的类型信息了。

在这种情况下，我们需要显式地指定类型信息，否则输出将被视为 Object 类型，这会导致低效的序列化。

// flatMap 使用 Lambda 表达式，必须通过 returns 明确声明返回类型 
DataStream<String> stream2 = clicks.flatMap((Event event, Collector<String> 
out) -> { 
out.collect(event.url); 
}).returns(Types.STRING); stream2.print();

当使用 map() 函数返回 Flink 自定义的元组类型时也会发生类似的问题。下例中的函数签名 Tuple2 map(Event value) 被类型擦除为 Tuple2 map(Event value)。

一般来说，这个问题可以通过多种方式解决：

import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.common.typeinfo.Types; 
import org.apache.flink.api.java.tuple.Tuple2; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class ReturnTypeResolve { 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) ); // 想要转换成二元组类型，需要进行以下处理 
96 // 1) 使用显式的 ".returns(...)" DataStream<Tuple2<String, Long>> stream3 = clicks .map( event -> Tuple2.of(event.user, 1L) ) .returns(Types.TUPLE(Types.STRING, Types.LONG)); stream3.print(); // 2) 使用类来替代 Lambda 表达式 clicks.map(new MyTuple2Mapper()) .print(); // 3) 使用匿名类来代替 Lambda 表达式 clicks.map(new MapFunction<Event, Tuple2<String, Long>>() { @Override public Tuple2<String, Long> map(Event value) throws Exception { return Tuple2.of(value.user, 1L); } }).print(); env.execute(); } // 自定义 MapFunction 的实现类 public static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String, 
Long>>{ @Override public Tuple2<String, Long> map(Event value) throws Exception { return Tuple2.of(value.user, 1L); } } 
}

3. 富函数类（Rich Function Classes）

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其

Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction 等。

既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

注：生命周期的概念在编程中其实非常重要，到处都有体现在 JVM 中，虚拟机会自动帮助我们管理对象的生命周期。Rich Function 有生命周期的概念。典型的生命周期方法有：

⚫ open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调用。所以像文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在 open()方法中完成。。

⚫ close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。

需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如 RichMapFunction 中的 map()，在每条数据到来后都会触发一次调用。

package com.atguigu.chapter05;import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class TransFromRichFunctionTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);DataStreamSource<Event> clicks = env.fromElements(new Event("Mary", "./home", 1000L),new Event("Bob", "./cart", 2000L),new Event("Alice", "./prod?id=1", 5 * 1000L),new Event("Cary", "./home", 60 * 1000L));// 将点击事件转换成长整型的时间戳输出clicks.map(new RichMapFunction<Event, Long>() {@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);System.out.println(" open生命周期被调用：" +getRuntimeContext().getIndexOfThisSubtask() + "号任务启动");}@Overridepublic Long map(Event value) throws Exception {return value.timestamp;}@Overridepublic void close() throws Exception {super.close();System.out.println("close生命周期被调用" +getRuntimeContext().getIndexOfThisSubtask() + "号任务结束");}}).print();env.execute();}
}

一个常见的应用场景就是，如果我们希望连接到一个外部数据库进行读写操作，那么将连接操作放在 map()中显然不是个好选择——因为每来一条数据就会重新连接一次数据库；所以我们可以在 open()中建立连接，在 map()中读写数据，而在 close()中关闭连接。所以我们推荐的最佳实践如下：

public class MyFlatMap extends RichFlatMapFunction<IN, OUT>> { @Override public void open(Configuration configuration) { // 做一些初始化工作 // 例如建立一个和 MySQL 的连接 } @Override public void flatMap(IN in, Collector<OUT out) { // 对数据库进行读写 } @Override public void close() { // 清理工作，关闭和 MySQL 数据库的连接。 
99 } 
}

富函数类提供了 getRuntimeContext()方法，可以获取到运行时上下文的一些信息，例如程序执行的并行度，任务名称，以及状态（state）。这使得我们可以大大扩展程序的功能，特别是对于状态的操作，使得 Flink 中的算子具备了处理复杂业务的能力。