Flink物理分区（Physical Partitioning）（出自b站尚硅谷）

前言

顾名思义，“分区”（partitioning）操作就是要将数据进行重新分布，传递到不同的流分区去进行下一步处理。其实我们对分区操作并不陌生，前面介绍聚合算子时，已经提到了 keyBy，它就是一种按照键的哈希值来进行重新分区的操作。只不过这种分区操作只能保证把数据按key“分开”，至于分得均不均匀、每个 key 的数据具体会分到哪一区去，这些是完全无从控制的——所以我们有时也说，keyBy 是一种逻辑分区（logical partitioning）操作。

如果说 keyBy 这种逻辑分区是一种“软分区”，那真正硬核的分区就应该是所谓的“物理分区”（physical partitioning）。也就是我们要真正控制分区策略，精准地调配数据，告诉每个数据到底去哪里。其实这种分区方式在一些情况下已经在发生了：例如我们编写的程序可能对多个处理任务设置了不同的并行度，那么当数据执行的上下游任务并行度变化时，数据就不应该还在当前分区以直通（forward）方式传输了——因为如果并行度变小，当前分区可能没有下游任务了；而如果并行度变大，所有数据还在原先的分区处理就会导致资源的浪费。所以这种情况下，系统会自动地将数据均匀地发往下游所有的并行任务，保证各个分区的负载均衡。

有些时候，我们还需要手动控制数据分区分配策略。比如当发生数据倾斜的时候，系统无法自动调整，这时就需要我们重新进行负载均衡，将数据流较为平均地发送到下游任务操作分区中去。Flink 对于经过转换操作之后的 DataStream，提供了一系列的底层操作接口，能够帮我们实现数据流的手动重分区。为了同 keyBy 相区别，我们把这些操作统称为“物理分区”操作。物理分区与 keyBy 另一大区别在于，keyBy 之后得到的是一个 KeyedStream，而物理分区之后结果仍是 DataStream，且流中元素数据类型保持不变。从这一点也可以看出，分区算子并不对数据进行转换处理，只是定义了数据的传输方式。

常见的物理分区策略有随机分配（Random）、轮询分配（Round-Robin）、重缩放（Rescale）和广播（Broadcast），下边我们分别来做了解。

1. 随机分区（shuffle）

最简单的重分区方式就是直接“洗牌”。通过调用 DataStream 的.shuffle()方法，将数据随机地分配到下游算子的并行任务中去。

随机分区服从均匀分布（uniform distribution），所以可以把流中的数据随机打乱，均匀地传递到下游任务分区，如图 5-9 所示。因为是完全随机的，所以对于同样的输入数据, 每次执行得到的结果也不会相同。

经过随机分区之后，得到的依然是一个 DataStream。

我们可以做个简单测试：将数据读入之后直接打印到控制台，将输出的并行度设置为 4，
中间经历一次 shuffle。执行多次，观察结果是否相同。

public class ShuffleTest {
   
	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