4.1.14 Flink-流处理框架-Flink流处理API之数据重分区操作

待我称王封你为后i 2022-11-04 11:24 217阅读 0赞

**目录**

1.写在前面

2.并行度

3.数据重分布

3.1 shuffle

3.2 rebalance与rescale

3.3 broadcast

3.4 global

3.5 partitionCustom

--------------------

# 1.写在前面 #

Flink的Transformation转换主要包括四种：**单数据流基本转换、基于Key的分组转换、多数据流转换和数据重分布转换**。

# 2.并行度 #

Flink使用并行度来定义某个算子被切分为多少个算子子任务。我们编写的大部分Transformation转换操作能够形成一个逻辑视图，当实际运行时，逻辑视图中的算子会被并行切分为一到多个算子子任务，每个算子子任务处理一部分数据。如下图所示，各个算子并行地在多个子任务上执行，假如算子的并行度为2，那么它有两个实例。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

并行度可以在一个Flink作业的执行环境层面统一设置，这样将设置该作业所有算子并行度，也可以对某个算子单独设置其并行度。如果不进行任何设置，默认情况下，一个作业所有算子的并行度会依赖于这个作业的执行环境。如果一个作业在本地执行，那么并行度默认是本机CPU核心数。当我们将作业提交到Flink集群时，需要使用提交作业的客户端，并指定一系列参数，其中一个参数就是并行度。

下面的代码展示了如何获取执行环境的默认并行度，如何更改执行环境的并行度。

val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 获取当前执行环境的默认并行度
    val defaultParallelism = senv.getParallelism
    // 设置所有算子的并行度为4，表示所有算子的并行执行的实例数为4
    senv.setParallelism(4)

也可以对某个算子设置并行度：

dataStream.map(new MyMapper).setParallelism(defaultParallelism * 2)

# 3.数据重分布 #

默认情况下，数据是自动分配到多个实例上的。有的时候，我们需要手动对数据在多个实例上进行分配，例如，我们知道某个实例上的数据过多，其他实例上的数据稀疏，产生了数据倾斜，这时我们需要将数据均匀分布到各个实例上，以避免部分实例负载过重。数据倾斜问题会导致整个作业的计算时间过长或者内存不足等问题。

下文涉及到的各个数据重分布算子的输入是`DataStream`，输出也是`DataStream`。`keyBy`也有对数据进行分组和数据重分布的功能，但`keyBy`输出的是`KeyedStream`。

## 3.1 shuffle ##

`shuffle`基于正态分布，将数据随机分配到下游各算子实例上。

dataStream.shuffle()

## 3.2 rebalance与rescale ##

`rebalance`使用Round-ribon思想将数据均匀分配到各实例上。Round-ribon是负载均衡领域经常使用的均匀分配的方法，上游的数据会轮询式地分配到下游的所有的实例上。如下图所示，上游的算子会将数据依次发送给下游所有算子实例。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

dataStream.rebalance()

`rescale`与`rebalance`很像，也是将数据均匀分布到各下游各实例上，但它的传输开销更小，因为`rescale`并不是将每个数据轮询地发送给下游每个实例，而是就近发送给下游实例。

dataStream.rescale()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

如上图所示，当上游有两个实例时，上游第一个实例将数据发送给下游第一个和第二个实例，上游第二个实例将数据发送给下游第三个和第四个实例，相比`rebalance`将数据发送给下游每个实例，`rescale`的传输开销更小。下图则展示了当上游有四个实例，上游前两个实例将数据发送给下游第一个实例，上游后两个实例将数据发送给下游第二个实例。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

## 3.3 broadcast ##

英文单词"broadcast"翻译过来为广播，在Flink里，数据会被复制并广播发送给下游的所有实例上。

dataStream.broadcast()

## 3.4 global ##

`global`会所有数据发送给下游算子的第一个实例上，使用这个算子时要小心，以免造成严重的性能问题。

dataStream.global()

## 3.5 partitionCustom ##

我们也可以使用`partitionCustom`来自**定义数据重分布逻辑**。**`partitionCustom`有两个参数：第一个参数是自定义的`Partitioner`，我们需要重写里面的`partition`函数；第二个参数是对数据流哪个字段使用`partiton`逻辑。`partition`函数的返回一个整数，表示该元素将被路由到下游第几个实例**。

`Partitioner[T]`中泛型T为指定的字段类型，比如我们要对`case class (id: Long, name: String, score: Double)`这个数据结构按照id均匀分配到下游各实例，那么泛型T就为id的数据类型Long。同时，泛型T也是`partition(key, numPartitions)`函数的第一个参数的数据类型。在调用`partitionCustom(partitioner， field)`时，第一个参数是我们重写的`Partitioner`，第二个参数表示按照id字段进行处理。

下面的代码按照数据流中的第二个字段进行数据重分布，当该字段中包含数字时，将被路由到下游算子的前半部分，否则被路由到后半部分。如果设置并行度为4，表示所有算子的实例数为4，或者说有4个分区，那么如果字符串包含数字时，该元素将被分配到第0个和第1个实例上，否则被分配到第2个和第3个实例上。

package com.flink.tutorials.api.transformations
    import org.apache.flink.api.common.functions.Partitioner
    import org.apache.flink.streaming.api.scala._
    object PartitionCustomExample {
      /**
        * Partitioner[T] 其中泛型T为指定的字段类型
        * 重写partiton函数，并根据T字段对数据流中的所有元素进行数据重分配
        * */
      class MyPartitioner extends Partitioner[String] {
        val rand = scala.util.Random
        /**
          * key 泛型T 即根据哪个字段进行数据重分配，本例中是(Int, String)中的String
          * numPartitons 为当前有多少个并行实例
          * 函数返回值是一个Int 为该元素将被发送给下游第几个实例
          * */
        override def partition(key: String, numPartitions: Int): Int = {
          var randomNum = rand.nextInt(numPartitions / 2)
          // 如果字符串中包含数字，该元素将被路由到前半部分，否则将被路由到后半部分。
          if (key.exists(_.isDigit)) {
            return randomNum
          } else {
            return randomNum + numPartitions / 2
          }
        }
      }
      def main(args: Array[String]): Unit = {
        val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        // 获取当前执行环境的默认并行度
        val defaultParalleism = senv.getParallelism
        // 设置所有算子的并行度为4，表示所有算子的并行执行的实例数为4
        senv.setParallelism(4)
        val dataStream: DataStream[(Int, String)] = senv.fromElements((1, "123"), (2, "abc"), (3, "256"), (4, "zyx")
          , (5, "bcd"), (6, "666"))
        // 对(Int, String)中的第二个字段使用 MyPartitioner 中的重分布逻辑
        val partitioned = dataStream.partitionCustom(new MyPartitioner, 1)
        partitioned.print()
        senv.execute("partition custom transformation")
      }
    }

package com.atguigu.apitest.transform;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class TransformTest6_Partition {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(4);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            dataStream.print("input");
    
            // 1. shuffle
            DataStream<String> shuffleStream = inputStream.shuffle();
    
    //        shuffleStream.print("shuffle");
    
            // 2. keyBy
    
    //        dataStream.keyBy("id").print("keyBy");
    
            // 3. global
            dataStream.global().print("global");
    
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/0262e1f7d3c340e4a7e6e23bd231b469.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/6ed75d930f4a4394b0c8f4c38ca4a4b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/4851a654e4ff4629bedbc406a678eca6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221023/baa68f3797cb4154b69e27159bad0c1c.png