物理分区操作

物理分区(physica1partitioning)操作的作用是根据指定的分区策略将数据重新分限到不同节点的 Task 实例上执行。当使用DataSteam提供的 API对数据处理过程中，赖于算子本身对数据的分区控制，如果用户希望自己控制数据分区，例如当数据发生了数据倾斜的时候，就需要通过定义物理分区策略的方式对数据集进行重新分布。

Flink 中已经提供了常见的分区策略，例如==随机分区(Random Partitioning)、平衡分区(Roundobin partitioning)、按比例分区(Roundrobin Partitioning)==等。如果给定的分队策略无法满足需求，也可以根据Flink提供的分区控制接口创建分区器，实现自定义区控制。

Flink内部提供的常见数据重分区策略如下所述：

• 随机分区(Random Partitioning):[DataStream ->DataStream]通过随机的方式将数据分配在下游算子的每个分区中，分区相对均衡，但是较容易失去原有数据的分区结构。
  //通过调用DatastreamAPI中的shuffle方法实现数据集的随机分区
val shuffleStream =dataStream.shuffle

• Roundrobin Partitioning:[DataStream ->DataStream]通过循环的方式对数据集中的数据进行重分区，能够尽可能保证每个分区的数据平当数据集发生数据倾斜的时候使用这种策略就是比较有效的优化方法。衡，
  //通过调用Datastream API中rebalance()方法实现数据的重平衡分区
val shuffleStream =dataStream.rebalance();

• Rescaling Partitioning: [DataStream ->DataStream]和 Roundrobin Partitioning一样，Rescaling Partitioning也是一种通过循环的方式进行数据重平衡的分区策略。但是不同的是，当使用Roundrobin Partitioning时，数据会全局性地通过网络介质传输到其他的节点完成数据的重新平衡，而Rescaling Partitioning仅仅会对上下游继承的算子数据进行重平衡，具体的分区主要根据上下游算子的并行度决定。例如上游算子的并发度为2，下游算子的并发度为4，就会发生上游算子中一个分区的数据按照同等比例将数据路由在下游的固定的两个分区中，另外一个分区同理路由到下游两个分区中。
  val shuffleStream =dataStream.rescale()

• 广播操作(Broadcasting):[DataStream ->DataStream
  广播策略将输入的数据集复制到下游算子的并行的Tasks实例中，tasks直接从本地内存中获取广播数据集，不再依赖于网络传输。这种分区策略适用小数据集，例如当大数据集关联小数据集时，可以通过广播的方式将小数据集分发制算子的每个分区中。
  //可以通过调用 DataStream API 的broadcast()方法实现广播分区
val shuffleStream = dataStream.broadcast();

• 自定义分区(Custom Partitioning):[DataStream ->DataStream除了使用已有的分区器之外，用户也可以实现自定义分区器，然后调用DatasueamAP]上partitionCustom()方法将创建的分区器应用到数据集上。如以下代码所示自定义分区器代码实现了当字段中包含“fink”关键字的数据放在partition为0的分区中,其余数据随机进行分区的策略，其中numPartitions是从系统中获取的并行度参数。

object customPartitioner extends Partitioner[String]{//获取随机数生成器
val r=scala.util.Random
override def partition(key: String,numPartitions: Int):Int ={
//定义分区策略，key中如果包含a则放在0分区中，其他情况则根据partitionsnum随机分区
if (key.contains("flink")) 0  else r.nextInt(numPartitions)}
}

自定义分区器定义好之后就可以调用DataSteamAPI的 partitionCustom来应用分区器，第二个参数指定分区器使用到的字段，对于Tuple类型数据，分区字段可以通过字段名称指定，其他类型数据集则通过位置索引指定。

//通过数据集字段名称指定分区字段
dataStream.partitioncustom(custompartitioner,"filed name");
11通过数据集字段索引指定分区字段
datastream,partitioncustom(custompartitioner,0)；