Flink实操 : 算子操作

野性酷女 2022-11-04 12:26 197阅读 0赞

### . ###

*  一 .前言
 *  二 .算子操作
 *   *  2.1. map
     *  2.2. flatMap
     *  2.3. mapPartition
     *  2.4. filter
     *  2.5. reduce/groupBy
     *  2.6. reduceGroup/groupBy
     *  2.7. aggregate
     *  2.8. join
     *  2.9. union
     *  2.10. rebalance
     *  2.11. hashPartition
     *  2.12. sortPartition
     *  2.13. keyBy
     *  2.14. Connect

# 一 .前言 #

<table> 
 <thead> 
  <tr> 
   <th align="left">Transformation</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">map</td> 
   <td align="left">将DataStream中的每一个元素转换为另外一个元素</td> 
  </tr> 
  <tr> 
   <td align="left">flatMap</td> 
   <td align="left">将DataStream中的每一个元素转换为0…n个元素</td> 
  </tr> 
  <tr> 
   <td align="left">mapPartition</td> 
   <td align="left">将一个分区中的元素转换为另一个元素</td> 
  </tr> 
  <tr> 
   <td align="left">filter</td> 
   <td align="left">过滤出来一些符合条件的元素</td> 
  </tr> 
  <tr> 
   <td align="left">reduce</td> 
   <td align="left">可以对一个DataStream或者一个group来进行聚合计算，最终聚合成一个元素</td> 
  </tr> 
  <tr> 
   <td align="left">reduceGroup</td> 
   <td align="left">将一个DataStream或者一个group聚合成一个或多个元素</td> 
  </tr> 
  <tr> 
   <td align="left">aggregate</td> 
   <td align="left">按照内置的方式来进行聚合。例如：SUM/MIN/MAX…</td> 
  </tr> 
  <tr> 
   <td align="left">distinct</td> 
   <td align="left">去重</td> 
  </tr> 
  <tr> 
   <td align="left">join</td> 
   <td align="left">将两个DataStream按照一定条件连接到一起，形成新的DataStream</td> 
  </tr> 
  <tr> 
   <td align="left">union</td> 
   <td align="left">将两个DataStream取并集，并不会去重</td> 
  </tr> 
  <tr> 
   <td align="left">rebalance</td> 
   <td align="left">让每个分区的数据均匀分布，避免数据倾斜</td> 
  </tr> 
  <tr> 
   <td align="left">partitionByHash</td> 
   <td align="left">按照指定的key进行hash分区</td> 
  </tr> 
  <tr> 
   <td align="left">sortPartition</td> 
   <td align="left">指定字段对分区中的数据进行排序</td> 
  </tr> 
  <tr> 
   <td align="left">keyBy</td> 
   <td align="left">按照指定的key来进行分流。</td> 
  </tr> 
  <tr> 
   <td align="left">connect</td> 
   <td align="left">将两个DataStream组装成一个ConnectedStreams</td> 
  </tr> 
 </tbody> 
</table>

# 二 .算子操作 #

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

## 2.1. map ##

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

将DataStream中的每一个元素转换为另外一个元素

**示例**

使用map操作，将以下数据

"1,张三", "2,李四", "3,王五", "4,赵六"

转换为一个scala的样例类。

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  创建一个`User`样例类
4.  使用`map`操作执行转换
5.  打印测试

**参考代码**

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
    
    
    // 使用map操作，将以下数据
    //
    // "1,张三", "2,李四", "3,王五", "4,赵六"
    // 转换为一个scala的样例类。
    //
    // 步骤
    // 获取ExecutionEnvironment运行环境
    // 使用fromCollection构建数据源
    // 创建一个User样例类
    // 使用map操作执行转换
    // 打印测试
    
    object MapOp { 
    
      // User实体类
      case class User( id:String , name :String )
    
      def main(args : Array[String]) : Unit = { 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data : DataStream[String]  = env.fromCollection(Array("1,张三", "2,李四", "3,王五", "4,赵六"))
    
        // 3.处理数据
        data.map( text => { 
          val uStr:Array[String] = text.split(",")
          User(uStr(0),uStr(1))
        })
    
        // 3.打印输出
        data.print()
    
        // 4.执行任务
        env.execute()
    
    
      }
    
    }

## 2.2. flatMap ##

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

将DataStream中的每一个元素转换为0…n个元素

**示例**

分别将以下数据，转换成`国家`、`省份`、`城市`三个维度的数据。

将以下数据

张三,中国,江西省,南昌市
    李四,中国,河北省,石家庄市
    Tom,America,NewYork,Manhattan

转换为

(张三,中国)
    (张三,中国,江西省)
    (张三,中国,江西省,江西省)
    (李四,中国)
    (李四,中国,河北省)
    (李四,中国,河北省,河北省)
    (Tom,America)
    (Tom,America,NewYork)
    (Tom,America,NewYork,NewYork)

**思路**

*  以上数据为一条转换为三条，显然，应当使用`flatMap`来实现
 *  分别在`flatMap`函数中构建三个数据，并放入到一个列表中
    
    姓名, 国家  
    姓名, 国家省份  
    姓名, 国家省份城市

**步骤**

1.  构建批处理运行环境
2.  构建本地集合数据源
3.  使用`flatMap`将一条数据转换为三条数据
    
     *  使用逗号分隔字段
     *  分别构建国家、国家省份、国家省份城市三个元组
4.  打印输出

**参考代码**

package com.boyi.transform
    
    import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
      //
      // 分别将以下数据，转换成国家、省份、城市三个维度的数据。
      //
      // 将以下数据
      //
      // 张三,中国,江西省,南昌市
      // 李四,中国,河北省,石家庄市
      // Tom,America,NewYork,Manhattan
      // 转换为
      //
      // (张三,中国)
      // (张三,中国,江西省)
      // (张三,中国,江西省,江西省)
      // (李四,中国)
      // (李四,中国,河北省)
      // (李四,中国,河北省,河北省)
      // (Tom,America)
      // (Tom,America,NewYork)
      // (Tom,America,NewYork,NewYork)
    
    object FlatMapOp { 
    
    
    
      def main(args : Array[String]) : Unit = { 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data : DataStream[String] =  env.fromCollection(List("张三,中国,江西省,南昌市","李四,中国,河北省,石家庄市","Tom,America,NewYork,Manhattan"))
        // 3.处理数据
        val res = data.flatMap(text => { 
          val dataStr : Array[String] = text.split(",")
    
          List(
            dataStr(0)+","+dataStr(1),
            dataStr(0)+","+dataStr(1)+","+dataStr(2),
            dataStr(0)+","+dataStr(1)+","+dataStr(2)+","+dataStr(3),
          )
    
        })
        // 3.打印输出
        res.print()
    
        // 4.执行任务
        env.execute()
    
      }
    
    }

## 2.3. mapPartition ##

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

将一个`分区`中的元素转换为另一个元素

**示例**

使用mapPartition操作，将以下数据

"1,张三", "2,李四", "3,王五", "4,赵六"

转换为一个scala的样例类。

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  创建一个`User`样例类
4.  使用`mapPartition`操作执行转换
5.  打印测试

**参考代码**

import org.apache.flink.api.scala.{ DataSet, ExecutionEnvironment, createTypeInformation}
    
    object MapPartitionOp { 
    
      case class User(id:String,name:String)
    
      def main(args : Array[String]) : Unit ={ 
        // 1. 创建流处理环境
        val env : ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data: DataSet[String] = env.fromCollection(List("1,张三", "2,李四", "3,王五", "4,赵六"))
        // 3.处理数据
        val res : DataSet[User]  =  data.mapPartition(x=>{ 
          x.map(x =>{ 
            val fieldArr = x.split(",")
            User(fieldArr(0),fieldArr(1))
          })
        })
        // 4.打印输出
        res.print()
        
      }
    
    }

> `map`和`mapPartition`的效果是一样的，但如果在map的函数中，需要访问一些外部存储。例如：
> 
> `访问mysql数据库，需要打开连接`, 此时效率较低。而使用`mapPartition`可以有效减少连接数，提高效率

## 2.4. filter ##

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

`过滤出来`一些符合条件的元素

示例：

过滤出来以下以`h`开头的单词。

"hadoop", "hive", "spark", "flink"

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`filter`操作执行过滤
4.  打印测试

**参考代码**

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
    
    object FilterOp { 
    
      def main(args : Array[String]):Unit ={ 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
    
        // 2.用fromCollection创建DataStream(fromCollection)
        val data : DataStream[String] = env.fromCollection(List("hadoop", "hive", "spark", "flink"))
        // 3.处理数据
        val res = data.filter(x => { x.startsWith("h")})
        // 4.打印输出
        res.print()
    
        // 5.执行任务
        env.execute()
      }
    
    }

## 2.5. reduce/groupBy ##

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

可以对一个`dataset`或者一个`group`来进行聚合计算，最终聚合成一个元素

**示例1**

请将以下元组数据，使用`reduce`操作聚合成一个最终结果

("java" , 1) , ("java", 1) ,("java" , 1)

将上传元素数据转换为`("java",3)`

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`redice`执行聚合操作
4.  打印测试

**参考代码**

import org.apache.flink.api.scala.{ DataSet, ExecutionEnvironment, createTypeInformation}
    
    object ReduceOp { 
    
      def main(args : Array[String]):Unit ={ 
        // 1. 创建流处理环境
        val env : ExecutionEnvironment  = ExecutionEnvironment.getExecutionEnvironment
    
        // 2.用fromCollection创建DataStream(fromCollection)
        val data : DataSet[(String,Int)]  = env.fromCollection(List(("java" , 1) , ("java", 1) ,("java" , 1) ))
        // 3.处理数据
        val res : DataSet[(String,Int)]  = data.reduce((x1,x2)=>{ 
         
          (x1._1,x1._2+x2._2)
        })
        // 4.打印输出
        res.print()
      }
    
    }

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

**示例**2

请将以下元组数据，下按照单词使用`groupBy`进行分组，再使用`reduce`操作聚合成一个最终结果

("java" , 1) , ("java", 1) ,("scala" , 1)

转换为

("java", 2), ("scala", 1)

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`groupBy`按照单词进行分组
4.  使用`reduce`对每个分组进行统计
5.  打印测试

**参考代码**

import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    object GroupByOp { 
      def main(args : Array[String]):Unit ={ 
        // 1. 创建流处理环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data = env.fromCollection(List(("java" , 1) , ("java", 1) ,("scala" , 1)  ))
        // 3. 使用`groupBy`按照单词进行分组
        val groupData = data.groupBy(_._1)
        // 4. 使用`reduce`对每个分组进行统计
        val reduceData = groupData.reduce((x1,x2) =>{ 
          (x1._1, x1._2+x2._2)
        })
        // 5.执行任务/输出结果
        reduceData.print()
      }
    
    }

## 2.6. reduceGroup/groupBy ##

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

可以对一个DataStream或者一个group来进行聚合计算，最终聚合成一个元素

reduce和reduceGroup的`区别`

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

*  reduce是将数据一个个拉取到另外一个节点，然后再执行计算
 *  reduceGroup是先在每个group所在的节点上执行计算，然后再拉取

**示例**

请将以下元组数据，下按照单词使用`groupBy`进行分组，再使用`reduceGroup`操作进行单词计数

("java" , 1) , ("java", 1) ,("scala" , 1)

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`groupBy`按照单词进行分组
4.  使用`reduceGroup`对每个分组进行统计
5.  打印测试

**参考代码**

import org.apache.flink.api.scala.{ DataSet, ExecutionEnvironment, createTypeInformation}
    
    object ReduceGroupOp { 
      def main(args:Array[String]):Unit={ 
    
        // 1. 创建流处理环境
        val env : ExecutionEnvironment  = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data = env.fromCollection(List(("java" , 1) , ("java", 1) ,("scala" , 1)))
        // 3.使用`groupBy`按照单词进行分组
        val groupByData= data.groupBy(_._1)
    
        // 4.使用`reduceGroup`对每个分组进行统计
    // val reduceGroupData = groupByData.reduceGroup { iter =>{ 
    // iter.reduce{(wc1, wc2) => (wc1._1,wc1._2 + wc2._2)}
    // } }
    
        val reduceGroupData =  groupByData.reduceGroup(x=>{ 
          x.reduce((x1,x2)=>{ (x1._1,x1._2+x2._2)})
        })
    
        // 5.打印输出
        reduceGroupData.print()
        
      }
    }

## 2.7. aggregate ##

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

按照内置的方式来进行聚合, Aggregate只能作用于`元组`上。例如：SUM/MIN/MAX…

**示例**

请将以下元组数据，使用`aggregate`操作进行单词统计

("java" , 1) , ("java", 1) ,("scala" , 1)

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`groupBy`按照单词进行分组
4.  使用`aggregate`对每个分组进行`SUM`统计
5.  打印测试

**参考代码**

import org.apache.flink.api.java.aggregation.Aggregations
    import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    object AggregateOp { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data =  env.fromCollection(List(("java" , 1) , ("java", 1) ,("scala" , 1)))
        // 3. 使用`groupBy`按照单词进行分组
        val groupedDataStream = data.groupBy(0)
    
        // 4. 使用`aggregate`对每个分组进行`SUM`统计
        val resultDataStream = groupedDataStream.aggregate(Aggregations.SUM, 1)
    
        // 4.打印输出
        resultDataStream.print()
    
      }
    }

> **注意**
> 
> 要使用aggregate，只能使用字段索引名或索引名称来进行分组`groupBy(0)`，否则会报一下错误:
> 
>     Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.

4.8. distinct

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

去除重复的数据

**示例**

请将以下元组数据，使用`distinct`操作去除重复的单词

("java" , 1) , ("java", 2) ,("scala" , 1)

去重得到

("java", 1), ("scala", 1)

**步骤**

1.  获取`ExecutionEnvironment`运行环境
2.  使用`fromCollection`构建数据源
3.  使用`distinct`指定按照哪个字段来进行去重
4.  打印测试

**参考代码**

// 1. 获取`ExecutionEnvironment`运行环境
    val env = ExecutionEnvironment.getExecutionEnvironment
    
    // 2. 使用`fromCollection`构建数据源
    val wordcountDataStream = env.fromCollection(
        List(("java" , 1) , ("java", 1) ,("scala" , 1)  )
    )
    
    // 3. 使用`distinct`指定按照哪个字段来进行去重
    val resultDataStream = wordcountDataStream.distinct(0)
    
    // 4. 打印测试
    resultDataStream.print()

## 2.8. join ##

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

使用join可以将两个DataStream连接起来

示例：

在`资料\测试数据源`中，有两个csv文件，有一个为`score.csv`，一个为`subject.csv`，分别保存了成绩数据以及学科数据。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QOLoEB7Z-1614675144146)(assets/1554285790727.png)\]][img-QOLoEB7Z-1614675144146_assets_1554285790727.png]

需要将这两个数据连接到一起，然后打印出来。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

**步骤**

1.  分别将资料中的两个文件复制到项目中的`data/join/input`中
2.  构建批处理环境
3.  创建两个样例类
    
        * 学科Subject（学科ID、学科名字）
        * 成绩Score（唯一ID、学生姓名、学科ID、分数——Double类型）
4.  分别使用`readCsvFile`加载csv数据源，并制定泛型
5.  使用join连接两个DataStream，并使用`where`、`equalTo`方法设置关联条件
6.  打印关联后的数据源

**参考代码**

// 学科Subject（学科ID、学科名字）
    case class Subject(id:Int, name:String)
    
    // 成绩Score（唯一ID、学生姓名、学科ID、分数）
    case class Score(id:Int, name:String, subjectId:Int, score:Double)
    
    
    def main(args: Array[String]): Unit = { 
        // 1. 获取`ExecutionEnvironment`运行环境
        val env = ExecutionEnvironment.getExecutionEnvironment
    
        // 3. 分别使用`readCsvFile`加载csv数据源
        val scoreDataStream = env.readCsvFile[Score]("./data/join/input/score.csv")
        val subjectDataStream = env.readCsvFile[Subject]("./data/join/input/subject.csv")
    
        // 4. 使用join连接两个DataStream，并使用`where`、`equalTo`方法设置关联条件
        val joinedDataStream = scoreDataStream.join(subjectDataStream).where(2).equalTo(0)
    
        // 5. 打印关联后的数据源
        joinedDataStream.print()
    }

## 2.9. union ##

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

将两个DataStream取并集，不会去重。

**示例**

将以下数据进行取并集操作

数据集1

"hadoop", "hive", "flume"

数据集2

"hadoop", "hive", "spark"

**步骤**

1.  构建批处理运行环境
2.  使用`fromCollection`创建两个数据源
3.  使用`union`将两个数据源关联在一起
4.  打印测试

**参考代码**

import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    object UnionOp { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data1 = env.fromCollection(List("hadoop", "hive", "flume"))
    
        val data2 = env.fromCollection(List("hadoop", "hive", "spark"))
    
        // 3.处理数据
        val res = data1.union(data2)
        // 4.打印输出
        res.print()
    
      }
    }

## 2.10. rebalance ##

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

Flink也会产生`数据倾斜`的时候，例如：当前的数据量有10亿条，在处理过程就有可能发生如下状况：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4SBPJD76-1614675144155)(assets/1554292062536.png)\]][img-4SBPJD76-1614675144155_assets_1554292062536.png]  
`rebalance`会使用轮询的方式将数据均匀打散，这是处理数据倾斜最好的选择。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

**步骤**

1.  构建批处理运行环境
2.  使用`env.generateSequence`创建0-100的并行数据
3.  使用`fiter`过滤出来`大于8`的数字
4.  使用map操作传入`RichMapFunction`，将当前子任务的ID和数字构建成一个元组
    
        在RichMapFunction中可以使用`getRuntimeContext.getIndexOfThisSubtask`获取子任务序号
5.  打印测试

**参考代码**

import org.apache.flink.api.common.functions.RichMapFunction
    import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    
    object RebalanceOp { 
      def main(args: Array[String]): Unit = { 
        // 1. 获取`ExecutionEnvironment`运行环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(2)
    
        // 2. 使用`env.generateSequence`创建0-100的并行数据
        val numDataStream = env.generateSequence(0, 100)
    
        // 3. 使用`fiter`过滤出来`大于8`的数字
        val filterDataStream = numDataStream.filter(_ > 8)
    
        // 4. 是否设置rebalance
        // filterDataStream.rebalance()
    
        // 5. 使用map操作传入`RichMapFunction`，将当前子任务的ID和数字构建成一个元组
        val resultDataStream = filterDataStream.map(new RichMapFunction[Long, (Long, Long)] { 
          override def map(in: Long): (Long, Long) = { 
            (getRuntimeContext.getIndexOfThisSubtask, in)
          }
        })
    
        // 6. 打印测试
        resultDataStream.print()
      }
    }

上述代码没有加rebalance，通过观察，有可能会出现数据倾斜。

在filter计算完后，调用`rebalance`，这样，就会均匀地将数据分布到每一个分区中。

## 2.11. hashPartition ##

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

按照指定的key进行hash分区

**示例**

基于以下列表数据来创建数据源，并按照hashPartition进行分区，然后输出到文件。

List(1,1,1,1,1,1,1,2,2,2,2,2)

**步骤**

1.  构建批处理运行环境
2.  设置并行度为`2`
3.  使用`fromCollection`构建测试数据集
4.  使用`partitionByHash`按照字符串的hash进行分区
5.  调用`writeAsText`写入文件到`data/parition_output`目录中
6.  打印测试

**参考代码**

package com.boyi.transform
    
    import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    object HashPartitionOp { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env : ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
        // 设置并行度为`2`
        env.setParallelism(2)
    
        // 2.用fromCollection创建DataStream(fromCollection)
        val data = env.fromCollection(List(1,1,1,1,1,1,1,2,2,2,2,2))
        // 3.处理数据
        val partionData = data.partitionByHash(_.toString)
    
        // 4. 调用`writeAsText`写入文件到`/opt/a/tmp/testPartion`目录中
        partionData.writeAsText("/opt/a/tmp/testPartion")
    
    
        // 5.打印输出
        partionData.print()
    
      }
    }

## 2.12. sortPartition ##

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

指定字段对分区中的数据进行`排序`

**示例**

按照以下列表来创建数据集

List("hadoop", "hadoop", "hadoop", "hive", "hive", "spark", "spark", "flink")

对分区进行排序后，输出到文件。

**步骤**

1.  构建批处理运行环境
2.  使用`fromCollection`构建测试数据集
3.  设置数据集的并行度为`2`
4.  使用`sortPartition`按照字符串进行降序排序
5.  调用`writeAsText`写入文件到`data/sort_output`目录中
6.  启动执行

**参考代码**

import org.apache.flink.api.common.operators.Order
    import org.apache.flink.api.scala.{ ExecutionEnvironment, createTypeInformation}
    
    object SortPartionOp { 
    
    
    
      def main(args : Array[String]) : Unit = { 
        // 1. 创建流处理环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        // 2.用fromCollection创建DataStream(fromCollection)
        val data = env.fromCollection(List("hadoop", "hadoop", "hadoop", "hive", "hive", "spark", "spark", "flink"))
        // 3.处理数据
        val res = data.sortPartition(_.toString,Order.DESCENDING)
        // 4.打印输出
        res.print()
      }
    
    }

## 2.13. keyBy ##

按照指定的key来进行分流。可以按照索引名/字段名来指定分组的字段.

**示例**

读取socket数据源, 进行单词的计数

**开发步骤**

1.  获取流处理运行环境
2.  设置并行度
3.  获取数据源
4.  转换操作
    
    1.  以空白进行分割
    2.  给每个单词计数1
    3.  根据单词分组
    4.  求和
5.  打印到控制台
6.  执行任务

**代码**

*  写法1 :

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
    
    import scala.collection.mutable.ListBuffer
    
    // 读取socket数据源, 进行单词的计数
    object KeyBy02Op { 
    
    
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment(2)
        // 2.用fromCollection创建DataStream(fromCollection)
        val data : DataStream[String] = env.readTextFile("hdfs://h23:8020/tmp/test/score.csv")
        // 3.处理数据
        val res = data.flatMap(x => x.split(",")).map(x=>(x,1)).keyBy(x => { 
          x._1
        }).sum(1)
    
        // 3.打印输出
        res.print()
        // 4.执行任务
        env.execute()
      }
    
    }

*  写法二:

## 2.14. Connect ##

`Connect`用来将两个DataStream组装成一个`ConnectedStreams`。它用了两个泛型，即不要求两个dataStream的element是同一类型。这样我们就可以把不同的数据组装成同一个结构.

**示例**

读取两个不同类型的数据源，使用connect进行合并打印。

**开发步骤**

1.  创建流式处理环境
2.  添加两个自定义数据源
3.  使用connect合并两个数据流,创建ConnectedStreams对象
4.  遍历ConnectedStreams对象,转换为DataStream
5.  打印输出,设置并行度为1
6.  执行任务

**自定义数据源**

/** * 创建自定义并行度为1的source * 实现从1开始产生递增数字 */
    class MyLongSourceScala extends SourceFunction[Long] { 
      var count = 1L
      var isRunning = true
    
      override def run(ctx: SourceContext[Long]) = { 
        while (isRunning) { 
          ctx.collect(count)
          count += 1
          TimeUnit.SECONDS.sleep(1)
        }
      }
    
      override def cancel() = { 
        isRunning = false
      }
    }
    
    /** * 创建自定义并行度为1的source * 实现从1开始产生递增字符串 */
    class MyStringSourceScala extends SourceFunction[String] { 
      var count = 1L
      var isRunning = true
    
      override def run(ctx: SourceContext[String]) = { 
        while (isRunning) { 
          ctx.collect("str_" + count)
          count += 1
          TimeUnit.SECONDS.sleep(1)
        }
      }
    
      override def cancel() = { 
        isRunning = false
      }
    }

**代码**

object StreamingDemoConnectScala { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流式处理环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // 2. 添加两个自定义数据源
        val text1: DataStream[Long] = env.addSource(new MyLongSourceScala)
        val text2: DataStream[String] = env.addSource(new MyStringSourceScala)
        // 3. 使用connect合并两个数据流,创建ConnectedStreams对象
        val connectedStreams: ConnectedStreams[Long, String] = text1.connect(text2)
        // 4. 遍历ConnectedStreams对象,转换为DataStream
        val result: DataStream[Any] = connectedStreams.map(line1 => { 
          line1
        }, line2 => { 
          line2
        })
        // 5. 打印输出,设置并行度为1
        result.print().setParallelism(1)
        // 6. 执行任务
        env.execute("StreamingDemoWithMyNoParallelSourceScala")
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221023/c73d9497dcef497ea603eb40b8e2299e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221023/685e714af6014224a8d2a98706897a9c.png
[img-QOLoEB7Z-1614675144146_assets_1554285790727.png]: /images/20221023/b8f14d1fb16e488da70652697bf0eb45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221023/727f5a5a1f80460c892679e70ec3cb02.png
[img-4SBPJD76-1614675144155_assets_1554292062536.png]: /images/20221023/a1784b5e217140b79d2d7ff819932f5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221023/4526f88642a74865af91176c8fb1cca6.png