数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个 DataStream 转换为新的 DataStream。一个 Flink 程序的核心，其实就是所有的转换操作，它们决定了处理的业务逻辑。
基本转换算子
映射（map）
map 是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素。
这里我们把对象中用户名提取出来
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) ); /** * 将对象转化为单个用户名 */ SingleOutputStreamOperator<String> map = stream.map(new MapFunction<Event, String>() { @Override public String map(Event event) throws Exception { return event.user; } }); map.print(); env.execute(); }
打印结果
Mary Bob
过滤
filter 转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。
这里我们过滤出来姓名为Mary的数据
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) ); /** * 过滤出来名称为Mary的数据 */ SingleOutputStreamOperator<Event> filter = stream.filter(new FilterFunction<Event>() { @Override public boolean filter(Event event) throws Exception { return event.user.equals("Mary"); } }); filter.print(); env.execute(); }
扁平映射（flatMap）
flatMap 操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap 可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。
flatMap 操作会应用在每一个输入事件上面，FlatMapFunction 接口中定义了 flatMap 方法，用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回 0 个、1 个或多个结果数据。因此 flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。所以 flatMap 方法也可以实现 map 方法和 filter 方法的功能，当返回结果是 0 个的时候，就相当于对数据进行了过滤，当返回结果是 1 个的时候，相当于对数据进行了简单的转换操作。
当名称为Mary的时候输出名称，当名称为Bob的时候输出名称和url
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) ); SingleOutputStreamOperator<String> flatMap = stream.flatMap(new FlatMapFunction<Event, String>() { @Override public void flatMap(Event event, Collector<String> collector) throws Exception { if (event.user.equals("Mary")) { collector.collect(event.user); } else if (event.user.equals("Bob")) { collector.collect(event.user); collector.collect(event.url); } } }); flatMap.print(); env.execute(); }
聚合算子
按键分区（keyBy）
对于 Flink 而言，DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区；这个操作就是通过 keyBy 来完成的。
keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
基于不同的 key，流中的数据将被分配到不同的分区中去，这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot中进行处理了。
在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。所以这里 key 如果是 POJO 的话，必须要重写 hashCode()方法。
这里根据用户名进行分组
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) ); /** * 这里根据用户名进行分组 */ KeyedStream<Event, String> eventStringKeyedStream = stream.keyBy(new KeySelector<Event, String>() { @Override public String getKey(Event event) throws Exception { return event.user; } }); eventStringKeyedStream.print(); env.execute(); }
需要注意的是，keyBy 得到的结果将不再是 DataStream，而是会将 DataStream 转换为KeyedStream。KeyedStream 可以认为是“分区流”或者“键控流”，它是对 DataStream 按照key 的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定 key 的类型。
KeyedStream 也继承自 DataStream，所以基于它的操作也都归属于 DataStream API。但它跟之前的转换操作得到的 SingleOutputStreamOperator 不同，只是一个流的分区操作，并不是一个转换算子。KeyedStream 是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如 sum，reduce）；而且它可以将当前算子任务的状态（state）也按照 key 进行划分、限定为仅对当前 key 有效。
简单聚合
有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink 为我们内置实现了一些最基本、最简单的聚合 API，主要有以下几种：
sum()：在输入流上，对指定的字段做叠加求和的操作。
min()：在输入流上，对指定的字段求最小值。
max()：在输入流上，对指定的字段求最大值。
minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。
maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
这里就不做过多例子了
归约聚合（reduce）重点
如果说简单聚合是对一些特定统计需求的实现，那么 reduce 算子就是一个一般化的聚合统计操作了。从大名鼎鼎的 MapReduce 开始，我们对 reduce 操作就不陌生：它可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。
调用 KeyedStream 的 reduce 方法时，需要传入一个参数，实现 ReduceFunction 接口。接口在源码中的定义如下：
public interface ReduceFunction<T> extends Function, Serializable { T reduce(T value1, T value2) throws Exception; }
ReduceFunction 接口里需要实现 reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件；所以，对于一组数据，我们可以先取两个进行合并，然后再将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据，
这也就是 reduce“归约”的含义。在流处理的底层实现过程中，实际上是将中间“合并的结果”作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。
其实，reduce 的语义是针对列表进行规约操作，运算规则由 ReduceFunction 中的 reduce方法来定义，而在 ReduceFunction 内部会维护一个初始值为空的累加器，注意累加器的类型和输入元素的类型相同，当第一条元素到来时，累加器的值更新为第一条元素的值，当新的元素到来时，新元素会和累加器进行累加操作，这里的累加操作就是 reduce 函数定义的运算规则。然后将更新以后的累加器的值向下游输出。
我们将数据流按照用户 id 进行分区，然后用一个 reduce 算子实现 sum 的功能，统计每个用户访问的频次；进而将所有统计结果分到一组，用另一个 reduce 算子实现 maxBy 的功能，记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。
这里用到了自定义source
import org.apache.flink.streaming.api.functions.source.SourceFunction; import java.util.Calendar; import java.util.Random; public class ClickSource implements SourceFunction<Event> { // 声明一个布尔变量，作为控制数据生成的标识位 private Boolean running = true; public void run(SourceContext<Event> sourceContext) throws Exception { Random random = new Random(); // 在指定的数据集中随机选取数据 String[] users = {"Mary", "Alice", "Bob", "Cary"}; String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"}; while (running) { sourceContext.collect(new Event( users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], Calendar.getInstance().getTimeInMillis() )); // 隔 1 秒生成一个点击事件，方便观测 Thread.sleep(1000); } } public void cancel() { running = false; } }
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 的流上。
自定义函数(UDF)
很容易发现，这些接口有一个共同特点：全部都以算子操作名称 + Function 命名，例如源算子需要实现 SourceFunction 接口，map 算子需要实现 MapFunction 接口，reduce 算子需要实现 ReduceFunction 接口。而且查看源码会发现，它们都继承自 Function 接口;这个接口是空的，主要就是为了方便扩展为单一抽象方法（Single Abstract Method，SAM）接口，这就是我们所说的“函数接口”——比如 MapFunction 中需要实现一个 map()方法，ReductionFunction中需要实现一个 reduce()方法，它们都是 SAM 接口。我们知道，Java 8 新增的 Lambda 表达式就可以实现 SAM 接口；所以这样的好处就是，我们不仅可以通过自定义函数类或者匿名类来实现接口，也可以直接传入 Lambda 表达式。这就是所谓的用户自定义函数（user-defined function，UDF）。
函数类（Function Classes）
对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类，例如 MapFunction、FilterFunction、ReduceFunction 等。
所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。
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); } }
