匿名函数（Lambda）
匿名函数（Lambda 表达式）是 Java 8 引入的新特性，方便我们更加快速清晰地写代码。Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。
Flink 的所有算子都可以使用 Lambda 表达式的方式来进行编码，但是，当 Lambda 表达式使用 Java 的泛型时，我们需要显式的声明类型信息。
下例演示了如何使用 Lambda 表达式来实现一个简单的 map() 函数，我们使用 Lambda 表达式来计算输入的平方。在这里，我们不需要声明 map() 函数的输入 i 和输出参数的数据类型，因为 Java 编译器会对它们做出类型推断。
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 就无法自动推断输出的类型信息了。
例如：
// 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<String, Long> map(Event value) 被类型擦除为 Tuple2 map(Event value)。
//使用 map 函数也会出现类似问题，以下代码会报错 DataStream<Tuple2<String, Long>> stream3 = clicks.map( event -> Tuple2.of(event.user, 1L) ); stream3.print();
一般来说，这个问题可以通过多种方式解决：
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) ); // 想要转换成二元组类型，需要进行以下处理 // 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); } }
富函数类（Rich Function Classes）
“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction 等。
既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
Rich Function 有生命周期的概念。典型的生命周期方法有：
open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调用。所以像文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在 open()方法中完成。
close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。
需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如 RichMapFunction 中的 map()，在每条数据到来后都会触发一次调用。
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>() { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); System.out.println(" 索 引 为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始"); } @Override public Long map(Event value) throws Exception { return value.timestamp; } @Override public void close() throws Exception { super.close(); System.out.println(" 索 引 为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束"); } }) .print(); env.execute(); }
输出结果是：
索引为 0 的任务开始 索引为 1 的任务开始 1> 1000 2> 2000 2> 60000 1> 5000 索引为 0 的任务结束 索引为 1 的任务结束
物理分区（Physical Partitioning）
顾名思义，“分区”（partitioning）操作就是要将数据进行重新分布，传递到不同的流分区去进行下一步处理。其实我们对分区操作并不陌生，前面介绍聚合算子时，已经提到了 keyBy，它就是一种按照键的哈希值来进行重新分区的操作。只不过这种分区操作只能保证把数据按key“分开”，至于分得均不均匀、每个 key 的数据具体会分到哪一区去，这些是完全无从控制的——所以我们有时也说，keyBy 是一种逻辑分区（logical partitioning）操作。
如果说 keyBy 这种逻辑分区是一种“软分区”，那真正硬核的分区就应该是所谓的“物理分区”（physical partitioning）。也就是我们要真正控制分区策略，精准地调配数据，告诉每个数据到底去哪里。其实这种分区方式在一些情况下已经在发生了：例如我们编写的程序可能对多个处理任务设置了不同的并行度，那么当数据执行的上下游任务并行度变化时，数据就不应该还在当前分区以直通（forward）方式传输了——因为如果并行度变小，当前分区可能没有下游任务了；而如果并行度变大，所有数据还在原先的分区处理就会导致资源的浪费。所以这种情况下，系统会自动地将数据均匀地发往下游所有的并行任务，保证各个分区的负载均衡。
有些时候，我们还需要手动控制数据分区分配策略。比如当发生数据倾斜的时候，系统无法自动调整，这时就需要我们重新进行负载均衡，将数据流较为平均地发送到下游任务操作分区中去。Flink 对于经过转换操作之后的 DataStream，提供了一系列的底层操作接口，能够帮我们实现数据流的手动重分区。为了同 keyBy 相区别，我们把这些操作统称为“物理分区”操作。物理分区与 keyBy 另一大区别在于，keyBy 之后得到的是一个 KeyedStream，而物理分区之后结果仍是 DataStream，且流中元素数据类型保持不变。从这一点也可以看出，分区算子并不对数据进行转换处理，只是定义了数据的传输方式。
常见的物理分区策略有随机分配（Random）、轮询分配（Round-Robin）、重缩放（Rescale）和广播（Broadcast），下边我们分别来做了解。
随机分区（shuffle）
最简单的重分区方式就是直接“洗牌”。通过调用 DataStream 的.shuffle()方法，将数据随机地分配到下游算子的并行任务中去。
随机分区服从均匀分布（uniform distribution），所以可以把流中的数据随机打乱，均匀地传递到下游任务分区，
因为是完全随机的，所以对于同样的输入数据, 每次执行得到的结果也不会相同。
经过随机分区之后，得到的依然是一个 DataStream。
我们可以做个简单测试：将数据读入之后直接打印到控制台，将输出的并行度设置为 4，中间经历一次 shuffle。执行多次，观察结果是否相同。
public static void main(String[] args) throws Exception { // 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 读取数据源，并行度为 1 DataStreamSource<Event> stream = env.addSource(new ClickSource()); // 经洗牌后打印输出，并行度为 4 stream.shuffle().print("shuffle").setParallelism(4); env.execute(); }
可以得到如下形式的输出结果：
shuffle:1> Event{user='Bob', url='./cart', timestamp=...} shuffle:4> Event{user='Cary', url='./home', timestamp=...} shuffle:3> Event{user='Alice', url='./fav', timestamp=...} shuffle:4> Event{user='Cary', url='./cart', timestamp=...} shuffle:3> Event{user='Cary', url='./fav', timestamp=...} shuffle:1> Event{user='Cary', url='./home', timestamp=...} shuffle:2> Event{user='Mary', url='./home', timestamp=...} shuffle:1> Event{user='Bob', url='./fav', timestamp=...} shuffle:2> Event{user='Mary', url='./home', timestamp=...}
轮询分区（Round-Robin）
轮询也是一种常见的重分区方式。简单来说就是“发牌”，按照先后顺序将数据做依次分发
public static void main(String[] args) throws Exception { // 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 读取数据源，并行度为 1 DataStreamSource<Event> stream = env.addSource(new ClickSource()); // 经轮询重分区后打印输出，并行度为 4 stream.rebalance().print("rebalance").setParallelism(4); env.execute(); }
输出结果的形式如下所示，可以看到，数据被平均分配到所有并行任务中去了。
rebalance:2> Event{user='Cary', url='./fav', timestamp=...} rebalance:3> Event{user='Mary', url='./cart', timestamp=...} rebalance:4> Event{user='Mary', url='./fav', timestamp=...} rebalance:1> Event{user='Cary', url='./home', timestamp=...} rebalance:2> Event{user='Cary', url='./cart', timestamp=...} rebalance:3> Event{user='Alice', url='./prod?id=1', timestamp=...} rebalance:4> Event{user='Cary', url='./prod?id=2', timestamp=...} rebalance:1> Event{user='Bob', url='./prod?id=2', timestamp=...} rebalance:2> Event{user='Alice', url='./prod?id=1', timestamp=...} ...
重缩放分区（rescale）
重缩放分区和轮询分区非常相似。当调用 rescale()方法时，其实底层也是使用 Round-Robin算法进行轮询，但是只会将数据轮询发送到下游并行任务的一部分中。也就是说，“发牌人”如果有多个，那么 rebalance 的方式是每个发牌人都面向所有人发牌；而 rescale的做法是分成小团体，发牌人只给自己团体内的所有人轮流发牌。
当下游任务（数据接收方）的数量是上游任务（数据发送方）数量的整数倍时，rescale的效率明显会更高。比如当上游任务数量是 2，下游任务数量是 6 时，上游任务其中一个分区的数据就将会平均分配到下游任务的 3 个分区中。
由于 rebalance 是所有分区数据的“重新平衡”，当 TaskManager 数据量较多时，这种跨节点的网络传输必然影响效率；而如果我们配置的 task slot 数量合适，用 rescale 的方式进行“局部重缩放”，就可以让数据只在当前 TaskManager 的多个 slot 之间重新分配，从而避免了网络传输带来的损耗。
从底层实现上看，rebalance 和 rescale 的根本区别在于任务之间的连接机制不同。rebalance将会针对所有上游任务（发送数据方）和所有下游任务（接收数据方）之间建立通信通道，这是一个笛卡尔积的关系；而 rescale 仅仅针对每一个任务和下游对应的部分任务之间建立通信通道，节省了很多资源。
public static void main(String[] args) throws Exception { StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 这里使用了并行数据源的富函数版本 // 这样可以调用 getRuntimeContext 方法来获取运行时上下文的一些信息 env.addSource(new RichParallelSourceFunction<Integer>() { @Override public void run(SourceContext<Integer> sourceContext) throws Exception { for (int i = 0; i < 8; i++) { // 将奇数发送到索引为 1 的并行子任务 // 将偶数发送到索引为 0 的并行子任务 if ((i + 1) % 2 == getRuntimeContext().getIndexOfThisSubtask()) { sourceContext.collect(i + 1); } } } @Override public void cancel() { } }) .setParallelism(2) .rescale() .print().setParallelism(4); env.execute(); }
这里使用 rescale 方法，来做数据的分区，输出结果是：
4> 3 3> 1 1> 2 1> 6 3> 5 4> 7 2> 4 2> 8
广播（broadcast）
这种方式其实不应该叫做“重分区”，因为经过广播之后，数据会在不同的分区都保留一份，可能进行重复处理。可以通过调用 DataStream 的 broadcast()方法，将输入数据复制并发送到下游算子的所有并行任务中去。
public static void main(String[] args) throws Exception { // 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 读取数据源，并行度为 1 DataStreamSource<Event> stream = env.addSource(new ClickSource()); // 经广播后打印输出，并行度为 4 stream.broadcast().print("broadcast").setParallelism(4); env.execute(); }
broadcast:3> Event{user='Mary', url='./cart', timestamp=...} broadcast:1> Event{user='Mary', url='./cart', timestamp=...} broadcast:4> Event{user='Mary', url='./cart', timestamp=...} broadcast:2> Event{user='Mary', url='./cart', timestamp=...} broadcast:2> Event{user='Alice', url='./fav', timestamp=...} broadcast:1> Event{user='Alice', url='./fav', timestamp=...} broadcast:3> Event{user='Alice', url='./fav', timestamp=...} broadcast:4> Event{user='Alice', url='./fav', timestamp=...}
可以看到，数据被复制然后广播到了下游的所有并行任务中去了。
全局分区（global）
全局分区也是一种特殊的分区方式。这种做法非常极端，通过调用.global()方法，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了 1，所以使用这个操作需要非常谨慎，可能对程序造成很大的压力。
自定义分区（Custom）
当 Flink 提 供 的 所 有 分 区 策 略 都 不 能 满 足 用 户 的 需 求 时 ， 我 们 可 以 通 过 使 用partitionCustom()方法来自定义分区策略。
在调用时，方法需要传入两个参数，第一个是自定义分区器（Partitioner）对象，第二个是应用分区器的字段，它的指定方式与 keyBy 指定 key 基本一样：可以通过字段名称指定，也可以通过字段位置索引来指定，还可以实现一个 KeySelector。
我们可以对一组自然数按照奇偶性进行重分区。
public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 将自然数按照奇偶分区 env.fromElements(1, 2, 3, 4, 5, 6, 7, 8) .partitionCustom(new Partitioner<Integer>() { @Override public int partition(Integer key, int numPartitions) { return key % 2; } }, new KeySelector<Integer, Integer>() { @Override public Integer getKey(Integer value) throws Exception { return value; } }) .print().setParallelism(2); env.execute(); }
