Flink万物之中Transform算子一

一时失言乱红尘 2023-03-13 05:51 19阅读 0赞

# Flink 的Transform算子 #

在上一篇中我们讲完了获取Flink执行环境与定义某些Source的方法，那么接下来也就是重要的Flink Transform算子的讲解环节了！！！老规矩先上图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center]  
图中有很多算子，我们会讲解一部分，其他用到后会在教程中讲解

# 一、基础算子 #

## 1.1 map ##

### 1.1.1 作用 ###

map可以理解为映射，对每个元素进行一定的变换后，映射为另一个元素，也就是一对一的转化

### 1.1.2 图解 ###

![在这里插入图片描述][20200511185119515.png_pic_center]

### 1.1.3 代码演示 ###

val streamMap = stream.map { x => x * 2 }

## 1.2 flatMap ##

### 1.2.1 干什么的 ###

可以理解为将元素摊平，一句话来理解就是一行变多行(一个元素根据某个规则变成0个、1个或者是多个)

### 1.2.2 详解 ###

flatMap的函数签名：def flatMap\[A,B\](as: List\[A\])(f: A ⇒ List\[B\]): List\[B\]

flatMap(List(1,2,3))(i ⇒ List(i,i))
    结果是List(1,1,2,2,3,3)
     
    List("a b", "c d").flatMap(line ⇒ line.split(" "))
    结果是List(a, b, c, d)。

## 1.3 Filter ##

### 1.3.1 干什么的 ###

根据规则去过滤出需要的数据

### 1.3.2 图解 ###

![在这里插入图片描述][20200511190234268.png_pic_center]

### 1.3.3 代码 ###

val streamFilter = stream.filter{
        x => x == 1
    }

# 二、keyby、aggregation、reduce等算子 #

## 2.1 图解关系 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 1]

## 2.2 KeyBy ##

### 2.2.1 图解 ###

![在这里插入图片描述][20200511190511107.png_pic_center]

### 2.2.2 原理 ###

keyBy算子将DataStream转换成一个KeyedStream。KeyedStream是一种特殊的DataStream，事实上，KeyedStream继承了DataStream，DataStream的各元素随机分布在各Task Slot中，KeyedStream的各元素按照Key分组，分配到各Task Slot中。我们需要向keyBy算子传递一个参数，以告知Flink以什么字段作为Key进行分组。

### 2.2.3 使用 ###

使用数字位置来指定key:

val dataStream: DataStream[(Int, Double)] = senv.fromElements((1, 1.0), (2, 3.2), (1, 5.5), (3, 10.0), (3, 12.5))
    // 使用数字位置定义Key 按照第一个字段进行分组
    val keyedStream = dataStream.keyBy(0)

可以使用字段名来指定Key，比如StockPrice里的股票代号symbol：

val stockPriceStream: DataStream[StockPrice] = stockPriceRawStream.keyBy(_.symbol)

## 2.3 aggregation ##

### 2.3.1 详解 ###

常见的聚合操作有sum、max、min等，这些聚合操作统称为aggregation。aggregation需要一个参数来指定按照哪个字段进行聚合。跟keyBy相似，我们可以使用数字位置来指定对哪个字段进行聚合，也可以使用字段名。与批处理不同，这些聚合函数是对流数据进行聚合，流数据是依次进入Flink的，聚合操作是对之前流入的数据进行统计聚合。sum算子的功能对该字段进行加和，并将结果保存在该字段上。min操作无法确定其他字段的数值。

### 2.3.2 相关算子简介 ###

*  sum算子的功能对该字段进行加和，并将结果保存在该字段上

val tupleStream = senv.fromElements(
          (0, 0, 0), (0, 1, 1), (0, 2, 2),
          (1, 0, 6), (1, 1, 7), (1, 2, 8)
    )
    // 按第一个字段分组，对第二个字段求和，打印出来的结果如下：
    //  (0,0,0)
    //  (0,1,0)
    //  (0,3,0)
    //  (1,0,6)
    //  (1,1,6)
    //  (1,3,6)
    val sumStream = tupleStream.keyBy(0).sum(1).print()

*  max算子对该字段求最大值，并将结果保存在该字段上。对于其他字段，该操作并不能保证其数值。

// 按第一个字段分组，对第三个字段求最大值max，打印出来的结果如下：
    //  (0,0,0)
    //  (0,0,1)
    //  (0,0,2)
    //  (1,0,6)
    //  (1,0,7)
    //  (1,0,8)
    val maxStream = tupleStream.keyBy(0).max(2).print()

*  maxBy算子对该字段求最大值，maxBy与max的区别在于，maxBy同时保留其他字段的数值，即maxBy可以得到数据流中最大的元素。

// 按第一个字段分组，对第三个字段求最大值maxBy，打印出来的结果如下：
    //  (0,0,0)
    //  (0,1,1)
    //  (0,2,2)
    //  (1,0,6)
    //  (1,1,7)
    //  (1,2,8)
    val maxByStream = tupleStream.keyBy(0).maxBy(2).print()

# 2.3.3 一些注意事项 #

其实，这些aggregation操作里已经封装了状态数据，比如，sum算子内部记录了当前的和，max算子内部记录了当前的最大值。由于内部封装了状态数据，而且状态数据并不会被清理，因此一定要避免在一个无限数据流上使用aggregation。  
注意，对于一个KeyedStream,一次只能使用一个aggregation操作，无法链式使用多个。

# 2.4 reduce #

### 2.4.1 图解 ###

![在这里插入图片描述][20200511190959997.png_pic_center]

### 2.4.2 详解 ###

reduce在按照同一个Key分组的数据流上生效，它接受两个输入，生成一个输出，即两两合一地进行汇总操作，生成一个同类型的新元素。

case class Score(name: String, course: String, score: Int)
    val dataStream: DataStream[Score] = senv.fromElements(
      Score("Li", "English", 90), Score("Wang", "English", 88), Score("Li", "Math", 85),
      Score("Wang", "Math", 92), Score("Liu", "Math", 91), Score("Liu", "English", 87))
    class MyReduceFunction() extends ReduceFunction[Score] {
      // reduce 接受两个输入，生成一个同类型的新的输出
      override def reduce(s1: Score, s2: Score): Score = {
        Score(s1.name, "Sum", s1.score + s2.score)
      }
    }
    val sumReduceFunctionStream = dataStream
          .keyBy("name")
          .reduce(new MyReduceFunction)
    简洁表达：
    val sumLambdaStream = dataStream
          .keyBy("name")
          .reduce((s1, s2) => Score(s1.name, "Sum", s1.score + s2.score))

## 2.5 fold ##

### 2.5.1 详解与举例 ###

在keys化的数据流上滚动的折叠。将当前的数据元与最后折叠的值组合并发出新值。

举例需求：序列（1,2,3,4,5），发出序列“start-1”，“start-1-2”，“start-1- 2-3”,. ..
    
    DataStream<String> result = keyedStream.fold("start", new FoldFunction<Integer, String& gt;() {
     @Override public String fold(String current, Integer value) { 
    return current + "-" + value; 
    } 
    });

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center]: /images/20230312/f539f5f12fea4f0b9adb197d589081d0.png
[20200511185119515.png_pic_center]: /images/20230312/222f6a7c1902460ea4a4fa2e8190dd28.png
[20200511190234268.png_pic_center]: /images/20230312/68818a27b56b43f1a6c1c8a6a39fedb1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230312/d8fea194325c44bc8491197c0f37b958.png
[20200511190511107.png_pic_center]: /images/20230312/96607a0eb2194c5286179dc005fa6f09.png
[20200511190959997.png_pic_center]: /images/20230312/30c18abe505a4d9ca6f960675509af41.png