主頁 >  其他 > 快速入門Flink(5)——Flink中的17種TransFormAction算子（面試必問，建議收藏）

快速入門Flink(5)——Flink中的17種TransFormAction算子（面試必問，建議收藏）

2020-09-18 21:26:28 其他

在這里插入圖片描述

上篇博客給大家講解了Flink的入門及dataSource 點擊，本篇博客給講解下Flink的17種常用的算子，本篇博客比較長，耐心看完（注意：面試經常被問到，建議收藏，如要對你有幫助的話麻煩，點贊 關注 評論），Flink專欄

1、Map

需求： 將 DataSet 中的每一個元素轉換為另外一個元素
示例: 使用 map 操作，將以下資料 “1,張三”, “2,李四”, “3,王五”, “4,趙六” 轉換為一個 scala 的樣例類，
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境，
  2. 使用 FromCollection 構建資料源，
  3. 創建一個 User 樣例類
  4. 使用 Map 將資料轉化為樣例類
  5. 列印輸出

參考代碼

import org.apache.flink.api.scala._

/**
 * @author 需求： 使用Map將資料轉換成樣例類
 * @date 2020/9/8 23:26
 * @version 1.0
 */

object BatchMap {
  case class User(name:String,age:String)
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用fromElements
    val data = env.fromElements("張三,19", "李四,30", "劉恒,299")
    //3.使用map將資料封裝成樣例類
    val userDataSet = data.map(s => {
      User(s.split(",")(0), s.split(",")(1))
    })
    //4.將資料輸出
    userDataSet.print()
  }
}

2、FlatMap

需求： 將 DataSet 中的每一個元素轉換為 0…n 個元素
實體： 分別將以下資料，轉換成 國家 、省份 、城市 三個維度的資料，
將以下資料

張三,中國,江西省,南昌市
李四,中國,河北省,石家莊市
Tom,America,NewYork,Manhattan

轉換為

張三,中國
張三,中國江西省
張三,中國江西省南昌市

解題思路
以上資料為一條轉換為三條，顯然，應當使用 flatMap 來實作 分別在 flatMap 函式中構建三個資料，并放入到一個串列中
顯示結果

姓名, 國家
姓名, 國家省份
姓名, 國家省份城市

實作步驟：

  1. 構建批處理運行環境
  2. 構建本地集合資料源
  3. 使用 flatMap 將一條資料轉換為三條資料
    a. 使用逗號分隔欄位
    b. 分別構建國家、國家省份、國家省份城市三個元組
  4. 列印輸出

代碼實作:

import org.apache.flink.api.scala._

/**
 * @author 需求：
 *        將"張三,中國,江西省,南昌市",
 *         "李四,中國,河北省,石家莊市",
 *         "Tom,America,NewYork,Manhattan"
 *        轉換為：
 *        張三,中國
 *        張三,中國江西省
 *        張三,中國江西省南昌市
 * @date 2020/9/9 0:11
 * @version 1.0
 */
object BachFlatMap {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用fromCollection構建資料集
    val dataSource = env.fromCollection(List("張三,中國,江西省,南昌市", "李四,中國,河北省,石家莊市", "Tom,America,NewYork,Manhattan"))
    val flatMap: DataSet[((String, String), (String, String), (String, String))] = dataSource.flatMap(line => {
      val arr = line.split(",")
      List(
         ((arr(0), arr(1)),
          (arr(0), arr(1) + arr(2)),
          (arr(0), arr(1) + arr(2) + arr(3))))
    })
   flatMap.print()
  }
}

3、MapPartition

需求： 將一個磁區中的元素轉換為另一個元素
示例： 使用 mapPartition 操作，將以下資料 “1,張三”, “2,李四”, “3,王五”, “4,趙六” 轉換為一個 scala 的樣例類，
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 創建一個 User 樣例類
  4. 使用 mapPartition 操作執行轉換
  5. 列印測驗

代碼實作：

import org.apache.flink.api.scala._
/**
 * @author 需求:將一天磁區中的資料轉換為一個樣例類
 *          "1,張三", "2,李四", "3,王五", "4,趙六"
 * @date 2020/9/9 21:57
 * @version 1.0
 */
object BachMapPartition {
  case class User(id:String,name:String)
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用FromElements構建資料集
    val dataSource = env.fromElements("1,張三", "2,李四", "3,王五", "4,趙六")
    //3.資料處理
    val mapPartitionDS: DataSet[User] = dataSource.mapPartition(textPartition => {
      textPartition.map( x => {
        val arrs = x.split(",")
        User(arrs(0), arrs(1))
      })
    })
    //4.結果輸出
    mapPartitionDS.print()
  }
}

4、Filter

需求： 過濾出來 一些符合條件的元素
Filter作用： Filter 函式在實際生產中特別實用，資料處理階段可以過濾掉大部分不符合業務的內容， 可以極 大減輕整體 flink 的運算壓力
實體： 使用filter過濾掉大于10的數字
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 filter 操作執行過濾
  4. 列印測驗

參考代碼

import org.apache.flink.api.scala._
/**
 * @author 需求：使用filter過濾掉大于10的數字
 *         過濾出來 一些符合條件的元素 Filter 函式在實際生產中特別實用，資料處理階段可以過濾掉大部分不符合業務的內容，
 *         可以極 大減輕整體 flink 的運算壓力
 * @date 2020/9/9 22:35
 * @version 1.0
 */
object BachFilter {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建1-20的資料集
    val dataSource = env.generateSequence(1, 20)
    //3.處理資料
    val filter = dataSource.filter(_ < 10)
    //4.結果輸出
    filter.print()
  }
}

5、Reduce

需求： 可以對一個 dataset 或者一個 group 來進行聚合計算，最終聚合成一個元素
實體： 請將以下元組資料，使用 reduce 操作聚合成一個最終結果 (“java” , 1) , (“java”, 1) ,(“java” , 1) 將上傳元素資料轉換為 (“java”,3)
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 redice 執行聚合操作
  4. 列印測驗

參考代碼:

import org.apache.flink.api.scala._

/**
 * @author 需求：請將以下元組資料，使用 reduce 操作聚合成一個最終結果
 *         ("java" , 1) , ("java", 1) ,("java" , 1)
 * @date 2020/9/9 22:39
 * @version 1.0
 */
object BachReduce {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用fromElements 構建資料集
    val dataSource = env.fromElements(("java", 1), ("java", 1), ("java", 2))
    //3.資料處理（根據key進行分組）
    val values = dataSource.groupBy(_._1)
    //4.使用reduce進行合并
    val reduce = values.reduce((v1, v2) => (v1._1, v2._2 + v1._2))
    //4.結果輸出
    reduce.print()
  }
}

6、ReduceGroup

可以對一個 dataset 或者一個 group 來進行聚合計算，最終聚合成一個元素 reduce 和 reduceGroup 的 區別
在這里插入圖片描述
首先 groupBy 函式會將一個個的單詞進行分組，分組后的資料被 reduce 一個個的拉 取過來，這種方式如果資料量大的情況下，拉取的資料會非常多，增加了網路 IO，
reduceGroup 是 reduce 的一種優化方案； 它會先分組 reduce，然后在做整體的 reduce；這樣做的好處就是可以減少網路 IO，
示例： 請將以下元組資料，下按照單詞使用 groupBy 進行分組，再使用 reduceGroup 操作進行單 詞計數(“java” , 1) , (“java”, 1) ,(“scala” , 1)
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 groupBy 按照單詞進行分組
  4. 使用 reduceGroup 對每個分組進行統計
  5. 列印測驗

參考代碼

import org.apache.flink.api.scala._

/**
 * @author 請將以下元組資料，下按照單詞使用 groupBy 進行分組，再使用 reduceGroup 操作進行單 詞計數("java" , 1) , ("java", 1) ,("scala" , 1)
 * @date 2020/9/11 22:15
 * @version 1.0
 */
object  BachReduceGroup {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val source = env.fromElements(("java", 1), ("Flin", 1), ("大資料", 1), ("java", 2))
    //3.使用reduceGroup進行分組求和
    val result = source.groupBy(0).reduceGroup(group => (group.reduce((a, b) => (a._1, a._2 + b._2))))
    //4.輸出
    result.print()
  }
}

7、Aggregate（重點）

介紹： 按照內置的方式來進行聚合，例如：SUM/MIN/MAX…
示例： 請將以下元組資料，使用 aggregate 操作進行單詞統計 (“java”, 1), (“大資料”, 2), (“大資料”, 10)
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 groupBy 按照單詞進行分組
  4. 使用 aggregate 對每個分組進行 SUM 統計
  5. 列印測驗

參考代碼

import org.apache.flink.api.scala._
/**
 * @author 請將以下元組資料，使用 aggregate 操作進行單詞統計 ("java", 1), ("大資料", 2), ("大資料", 10)
 * @date 2020/9/11 22:30
 * @version 1.0
 */
object BachAggregate {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用fromElements構建資料集
    val sourceData = env.fromElements(("java", 1), ("大資料", 2), ("大資料", 10))
    //3.使用groupBy進行分組然后使用aggregate求出最大值
    val result = sourceData.groupBy(0).aggregate(Aggregations.MAX, 1)
    //4.結果輸出
    result.print()
  }
}

8、minBy 和 maxBy

介紹： 獲取指定欄位的最大值、最小值
示例： 請將以下元組資料，使用 aggregate 操作進行單詞統計 (1, “yuwen”, 89.0) , (2, “shuxue”, 92.2),(3, “yingyu”, 89.99)，(4, “wuli”, 98.9)， (1, “yuwen”, 88.88)，(1, “wuli”, 93.00)，(1, “yuwen”, 94.3)
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 groupBy 按照單詞進行分組
  4. 使用 maxBy、minBy對每個分組進行操作
  5. 列印測驗

參考代碼：

import org.apache.flink.api.scala._
import scala.collection.mutable
import scala.util.Random

/**
 * @author
 * @date 2020/9/11 22:40
 * @version 1.0
 */
object BachMinByAndMaxBy {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val data = new mutable.MutableList[(Int, String, Double)]
    data += ((1, "liuheng", 89.0))
    data += ((2, "shuxue", 92.2))
    data += ((3, "yingyu", 89.99))
    data += ((4, "wuli", 98.9))
    data += ((1, "yuwen", 88.88))
    data += ((1, "wuli", 93.00))
    data += ((1, "yuwen", 94.3))
    val sourceData = env.fromCollection(Random.shuffle(data))
    //3.使用MinBy求出最小值與MaxBy求出最大值
    val min = sourceData.groupBy(1).minBy(2)
    val max = sourceData.groupBy(1).maxBy(2)
    //4.輸出最小值
    min.print()
    println("-----------------------------")
    //5.輸出最大值
    max.print()
  }
}

9、Distinct

介紹： 去除重復的資料
示例： 請將以下元組資料，使用 distinct 操作去除重復的單詞 (“java” , 1) , (“java”, 1) ,(“scala” , 1) 去重得到 (“java”, 1), (“scala”, 1)
實作步驟：

  1. 獲取 ExecutionEnvironment 運行環境
  2. 使用 fromCollection 構建資料源
  3. 使用 distinct 指定按照哪個欄位來進行去重
  4. 列印測驗

參考代碼：

import org.apache.flink.api.scala._
/**
 * @author 需求：使用distinct求("java", 1), ("java", 2), ("scala", 1) 去掉重復的資料
 * @date 2020/9/12 22:56
 * @version 1.0
 */
object BachDistinct {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境(背景關系物件)
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.使用fromElements
    val dataSource = env.fromElements(("java", 1), ("java", 2), ("scala", 1))
    //3.使用distinct去掉重復的
    val distinct = dataSource.distinct(0)
    //4.結果輸出
    distinct.print()
  }
}

10、Join

介紹： 使用 join 可以將兩個 DataSet 連接起來
示例： 有兩個 csv 檔案，有一個為 score.csv ，一個為 subject.csv ，分 別保存了成績資料以及學科資料
sorce.csv

1,語數
2,英物
3,化生
4,文學
5,語理
6,學物

subject.csv

1,張三,1,98
2,張三,2,77.5
3,張三,3,89
4,張三,4,65
5,張三,5,78
6,張三,6,70
9,李四,3,65
10,李四,4,78
11,李四,5,70
12,李四,6,78
13,王五,1,70
14,王五,2,78

實作步驟：

  1. 分別將資料中的兩個檔案復制到專案中的 data/ 中
  2. 構建批處理環境
  3. 創建兩個樣例類
    a. 學科 Subject（學科 ID、學科名字）
    b. 成績 Score（唯一 ID、學生姓名、學科 ID、分數——Double 型別）
  4. 分別使用 readCsvFile 加載 csv 資料源，并制定泛型
  5. 使用 join 連接兩個 DataSet，并使用 where 、 equalTo 方法設定關聯條件
  6. 列印關聯后的資料源

參開代碼：

import org.apache.flink.api.scala._

/**
 * @author 需求：使用join的方式將sorce.csv檔案與subject.csv檔案進行關聯
 * @date 2020/9/12 23:38
 * @version 1.0
 */
object BachJoin {
  //構建樣例類
  case class sorce(id:String,subject:String)
  case class subject(id:String,name:String,sid:String,source:String)
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集(使用檔案方式)
    val sorce = env.readCsvFile[sorce]("./data/score.csv")
    val subject = env.readCsvFile[subject]("./data/subject.csv")
    //3.使用join將兩個檔案中的資料進行關聯
    val joinData = sorce.join(subject).where(_.id).equalTo(_.sid)
    //4.結果輸出
    joinData.print()
  }
}

11、LeftOuterJoin

介紹： 左外連接,左邊的 Dataset 中的每一個元素，去連接右邊的元素
示例： 請將以下元組資料

(用戶 id,用戶姓名)
(1, “zhangsan”) ,
(2, “lisi”) ,
(3 , “wangwu”) ,
(4 , “zhaoliu”)

元組資料

(用戶 id,所在城市)
(1, “beijing”),
(2, “shanghai”),
(4, “guangzhou”)

回傳如下資料：

(3,wangwu,null)
(1,zhangsan,beijing)
(2,lisi,shanghai)
(4,zhaoliu,guangzhou)

參考代碼


import scala.collection.mutable.ListBuffer

/**
 * @author 需求：使用左連接 請將以下元組資料(用戶 id,用戶姓名)
 *         (1, "zhangsan") ,
 *         (2, "lisi") ,
 *         (3 , "wangwu") ,
 *         (4 , "zhaoliu")
 *         元組資料
 *         (用戶 id,所在城市)
 *         (1, "beijing"),
 *         (2, "shanghai"),
 *         (4, "guangzhou")
 *         回傳如下資料：
 *         (3,wangwu,null)
 *         (1,zhangsan,beijing)
 *         (2,lisi,shanghai)
 *         (4,zhaoliu,guangzhou)
 * @date 2020/9/15 23:30
 * @version 1.0
 */
object BachLeftOuterJoin {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.創建要測驗的資料集
    val data1 = ListBuffer[Tuple2[Int, String]]()
    data1.append((1, "zhangsan"))
    data1.append((2, "lisi"))
    data1.append((3, "wangwu"))
    data1.append((4, "zhaoliu"))
    val data2 = ListBuffer[Tuple2[Int, String]]()
    data2.append((1, "beijing"))
    data2.append((2, "shanghai"))
    data2.append((4, "guangzhou"))
    //2.2 使用fromCollection構建資料集
    val test1 = env.fromCollection(data1)
    val test2 = env.fromCollection(data2)
    //3.使用leftOuterJoin 進行關聯
    val result = test1.leftOuterJoin(test2).where(0).equalTo(0).apply((first, second)=>{
      if (second==null){
        (first._1,first._2,"null")
      }else{
        (first._1,first._2,second._2)
      }
    })
    //4.結果輸出
    result.print()
  }
}

12、RightOuterJoin

實體： 右外連接,左邊的 Dataset 中的每一個元素，去連接左邊的元素
示例： 請將以下元組資料

(用戶 id,用戶姓名)
(1, “zhangsan”) ,
(2, “lisi”) ,
(3 , “wangwu”) ,
(4 , “zhaoliu”)

元組資料

(用戶 id,所在城市)
(1, “beijing”),
(2, “shanghai”),
(4, “guangzhou”)

回傳如下資料：

(1,zhangsan,beijing)
(2,lisi,shanghai)
(4,zhaoliu,guangzhou)

參考代碼


import org.apache.flink.api.scala._

import scala.collection.mutable.ListBuffer

/**
 * @author 需求：使用左連接 請將以下元組資料(用戶 id,用戶姓名)
 *         (1, "zhangsan") ,
 *         (2, "lisi") ,
 *         (3 , "wangwu") ,
 *         (4 , "zhaoliu")
 *         元組資料
 *         (用戶 id,所在城市)
 *         (1, "beijing"),
 *         (2, "shanghai"),
 *         (4, "guangzhou")
 *         回傳如下資料：
 *         (1,zhangsan,beijing)
 *         (4,zhaoliu,guangzhou)
 *         (2,lisi,shanghai)
 * @date 2020/9/15 23:30
 * @version 1.0
 */
object BachRightOuterJoin {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.創建要測驗的資料集
    val data1 = ListBuffer[Tuple2[Int, String]]()
    data1.append((1, "zhangsan"))
    data1.append((2, "lisi"))
    data1.append((3, "wangwu"))
    data1.append((4, "zhaoliu"))
    val data2 = ListBuffer[Tuple2[Int, String]]()
    data2.append((1, "beijing"))
    data2.append((2, "shanghai"))
    data2.append((4, "guangzhou"))
    //2.2 使用fromCollection構建資料集
    val test1 = env.fromCollection(data1)
    val test2 = env.fromCollection(data2)
    //3.使用rightOuterJoin 進行關聯
    val result = test1.rightOuterJoin(test2).where(0).equalTo(0).apply((first, second)=>{
      if (second==null){
        (first._1,first._2,"null")
      }else{
        (first._1,first._2,second._2)
      }
    })
    //4.結果輸出
    result.print()
  }
}

13、fullOuterJoin

介紹： 全外連接,左右兩邊的元素，全部連接
示例： 請將以下元組資料

(用戶 id,用戶姓名)
(1, “zhangsan”) ,
(2, “lisi”) ,
(3 , “wangwu”) ,
(4 , “zhaoliu”)

元組資料

(用戶 id,所在城市)
(1, “beijing”),
(2, “shanghai”),
(4, “guangzhou”)

回傳如下資料：

(3,wangwu,null)
(1,zhangsan,beijing)
(2,lisi,shanghai)
(4,zhaoliu,guangzhou)

擴展：

  • OPTIMIZER_CHOOSES： 將選擇權交予Flink優化器，相當于沒有給提示；
  • BROADCAST_HASH_FIRST：廣播第一個輸入端，同時基于它構建一個哈希表，而第 二個輸入端作為探索端，選擇這種策略的場景第一個輸入端規模很小；
  • BROADCAST_HASH_SECOND：廣播第二個輸入端并基于它構建哈希表，第一個輸入端 作為探索端，選擇這種策略的場景是第二個輸入端的規模很小；
  • REPARTITION_HASH_FIRST：該策略會導致兩個輸入端都會被重磁區，但會基于第 一個輸入端構建哈希表，該策略適用于第一個輸入端資料量小于第二個輸入端的資料量，但這 兩個輸入端的規模仍然很大，優化器也是當沒有辦法估算大小，沒有已 存在的磁區以及排序 順序可被使用時系統默認采用的策略；
  • REPARTITION_HASH_SECOND： 該策略會導致兩個輸入端都會被重磁區，但會基于 第二個輸入端構建哈希表，該策略適用于兩個輸入端的規模都很大，但第二個輸入端的資料量 小于第一個輸入端的情況；
  • REPARTITION_SORT_MERGE：輸入端被以流的形式進行連接并合并成排過序的輸入， 該策略適用于一個或兩個輸入端都已 排過序的情況；

參考代碼：

import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
import org.apache.flink.api.scala._

import scala.collection.mutable.ListBuffer

/**
 * @author 需求：
 *            請將以下元組資料(用戶 id,用戶姓名)
 *            (1, "zhangsan") ,
 *            (2, "lisi") ,
 *            (3 , "wangwu") ,
 *            (4 , "zhaoliu")
 *            元組資料(用戶 id,所在城市)
 *            (1, "beijing"),
 *            (2, "shanghai"),
 *            (4, "guangzhou")
 *            回傳如下資料：
 *            (3,wangwu,null)
 *            (1,zhangsan,beijing)
 *            (2,lisi,shanghai)
 *            (4,zhaoliu,guangzhou)
 * @date 2020/9/15 23:43
 * @version 1.0
 */
object BachFullOuterJoin {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.創建要測驗的資料集
    val data1 = ListBuffer[Tuple2[Int, String]]()
    data1.append((1, "zhangsan"))
    data1.append((2, "lisi"))
    data1.append((3, "wangwu"))
    data1.append((4, "zhaoliu"))
    val data2 = ListBuffer[Tuple2[Int, String]]()
    data2.append((1, "beijing"))
    data2.append((2, "shanghai"))
    data2.append((4, "guangzhou"))
    //2.2 使用fromCollection構建資料集
    val test1 = env.fromCollection(data1)
    val test2 = env.fromCollection(data2)
    //3.使用fullOuterJoin 進行關聯
    val result = test1.fullOuterJoin(test2,JoinHint.REPARTITION_SORT_MERGE).where(0).equalTo(0).apply((first, second)=>{
      if (first==null){
        (second._1,"null",second._2)
      } else if (second==null){
        (first._1,first._2,"null")
      }else{
        (first._1,first._2,second._2)
      }
    })
    //4.結果輸出
    result.print()
  }
}

14、cross

介紹： 和 join 類似，但是這種交叉操作會產生笛卡爾積，在資料比較大的時候，是非常消耗記憶體 的操作；
示例： 請將以下元組資料 (1, 4, 7), (2, 5, 8), (3, 6, 9)
元組資料 (10, 40, 70), (20, 50, 80), (30, 60, 90)

進行笛卡爾積，回傳如下資料：

Buffer(((1,4,7),(10,40,70)), ((1,4,7),(20,50,80)), ((1,4,7),(30,60,90)), ((2,5,8),(10,40,70)), ((2,5,8),(20,50,80)), ((2,5,8),(30,60,90)), ((3,6,9),(10,40,70)), ((3,6,9),(20,50,80)), ((3,6,9),(30,60,90)))

參考代碼：


import org.apache.flink.api.scala._

/**
 * @author
 * @date 2020/9/15 23:50
 * @version 1.0
 */
object BatchCross {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    println("==============================cross=======================================")
    cross(env)
    println("==============================crossWithTiny=======================================")
    crossWithTiny(env)
    println("==============================crossWithHuge=======================================")
    crossWithHuge(env)
  }

  /**
   * 笛卡爾集
   *
   * @param env
   */
  def cross(env: ExecutionEnvironment) = {
    //1.使用 fromElements定義兩個dataSet
    val data1 = env.fromElements((1, 4, 7), (2, 5, 8), (3, 6, 9))
    val data2 = env.fromElements((10, 40, 70), (20, 50, 80), (30, 60, 90))
    val result = data1.cross(data2)
    println(result.collect())
  }

  /**
   * 暗示第二個輸入較小的交叉
   *
   * @param env
   */
  def crossWithTiny(env: ExecutionEnvironment) = {
    //1.定義 case class
    case class Coord(id: Int, x: Int, y: Int)
    val data1: DataSet[Coord] = env.fromElements( Coord(2, 5, 8), Coord(1, 4, 7),Coord(3, 6, 9))
    val data2: DataSet[Coord] = env.fromElements( Coord(20, 50, 80),Coord(10, 40, 70), Coord(30, 60, 90))
    val result = data1.crossWithTiny(data2)
    result.print()
  }

  def crossWithHuge(env: ExecutionEnvironment) = {
    //1.定義 case class
    case class Coord(id: Int, x: Int, y: Int)
    val data1: DataSet[Coord] = env.fromElements(Coord(1, 4, 7), Coord(2, 5, 8), Coord(3, 6, 9))
    val data2: DataSet[Coord] = env.fromElements(Coord(10, 40, 70), Coord(20, 50, 80), Coord(30, 60, 90))
    val result = data1.crossWithHuge(data2)
    result.print()
  }
}

15、Union

介紹： 將多個 DataSet 合并成一個 DataSet【注意】：union 合并的 DataSet 的型別必須是一致 的
示例：
將以下資料進行取并集操作
資料集

1 “hadoop”, “hive”, “flume”

資料集 2

“hadoop”, “hive”, “spark”

實作步驟：

  1. 構建批處理運行環境
  2. 使用 fromCollection 創建兩個資料源
  3. 使用 union 將兩個資料源關聯在一起
  4. 列印測驗

參考代碼：

import org.apache.flink.api.scala._

/**
 * @author 需求：
 *         將以下資料進行取并集操作
 *         資料集
 *         1 "hadoop", "hive", "flume"
 *         資料集
 *         2 "hadoop", "hive", "spark
 * @date 2020/9/16 0:05
 * @version 1.0
 */
object BachUnion {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val data1 = env.fromCollection(List("hadoop", "hive", "flume"))
    val data2 = env.fromCollection(List("hadoop", "hive", "spark"))
    val result = data1.union(data2)
    result.print()
  }
}

16、Rebalance（重點）

介紹：
Flink 也有資料傾斜的時候，比如當前有資料量大概 10 億條資料需要處理，在處理程序中 可能會 發生如圖所示的狀況：
在這里插入圖片描述
這個時候本來總體資料量只需要 10 分鐘解決的問題，出現了資料傾斜，機器 1 上的 任務需 要 4 個小時才能完成，那么其他 3 臺機器執行完畢也要等待機器 1 執行完畢后才 算整體將任務完成；所以在實際的作業中，出現這種情況比較好的解決方案就是—rebalance（內 部使用 round robin 方法將資料均勻打散，這對于資料傾斜時是很 好的選擇，）
在這里插入圖片描述
實作步驟：

  1. 構建批處理運行環境
  2. 使用 env.generateSequence 創建 0-100 的并行資料
  3. 使用 fiter 過濾出來 大于 8 的數字
  4. 使用 map 操作傳入 RichMapFunction ，將當前子任務的 ID 和數字構建成一個元組
  5. 在 RichMapFunction 中可以使用 getRuntimeContext.getIndexOfThisSubtask 獲取子 任務序號
  6. 列印測驗

代碼實作：

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._

/**
 * @author 對資料集進行再平衡，重磁區，消除資料傾斜
 * @date 2020/9/16 0:21
 * @version 1.0
 */
object BatchRebalance {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val data = env.generateSequence(0, 100)
    val ds = data.filter(_ > 8)
    //3.對資料進行在平衡操作
    val value1 = ds.rebalance().map(new RichMapFunction[Long, (Int, Long)] {
      override def map(value: Long): (Int, Long) = {
        (getRuntimeContext.getIndexOfThisSubtask, value)
      }
    })
    //4.結果輸出
    value1.print()
  }
}

17、First（重點）

介紹： 根據給定的 key 對一個資料集取前 N 條資料（往往在公司中是經常用到了，比如頭條中的熱搜Top10）
實作步驟：

  1. 構建批處理運行環境
  2. 使用 fromCollection 構建測驗資料集
  3. 使用 first 獲取前 N 條資料
  4. 列印測驗

參考代碼：

import org.apache.flink.api.common.operators.Order
import org.apache.flink.api.scala._

import scala.collection.mutable.ListBuffer

/**
 * @author  需求：根據給定的 key 對一個資料集取前 N 條資料
 * @date 2020/9/16 19:07
 * @version 1.0
 */
object BachFirst {
  def main(args: Array[String]): Unit = {
    //1.構建運行環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val data = ListBuffer[Tuple2[Int,String]]()
    data.append((2,"zs"))
    data.append((4,"ls"))
    data.append((3,"ww"))
    data.append((1,"xw"))
    data.append((1,"aw"))
    data.append((1,"mw"))
    val text = env.fromCollection(data)
    //3.使用first去前三條資料
    val first = text.first(3)
    val sortFirst = text.sortPartition(0, Order.ASCENDING).sortPartition(1, Order.DESCENDING).first(3)
    //4.結果資料
    first.print()
    sortFirst.print()
  }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/76632.html

標籤：其他

上一篇：python青蛙爬井

下一篇：同心科赫曲線應該怎么弄？

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • 網閘典型架構簡述

    網閘架構一般分為兩種：三主機的三系統架構網閘和雙主機的2+1架構網閘。 三主機架構分別為內端機、外端機和仲裁機。三機無論從軟體和硬體上均各自獨立。首先從硬體上來看，三機都用各自獨立的主板、記憶體及存盤設備。從軟體上來看，三機有各自獨立的作業系統。這樣能達到完全的三機獨立。對于“2+1”系統，“2”分為 ......

    uj5u.com 2020-09-10 02:00:44 more
  • 如何從xshell上傳檔案到centos linux虛擬機里

    如何從xshell上傳檔案到centos linux虛擬機里及：虛擬機CentOs下執行 yum -y install lrzsz命令，出現錯誤：鏡像無法找到軟體包 前言 一、安裝lrzsz步驟 二、上傳檔案 三、遇到的問題及解決方案 總結 前言 提示：其實很簡單，往虛擬機上安裝一個上傳檔案的工具 ......

    uj5u.com 2020-09-10 02:00:47 more
  • 一、SQLMAP入門

    一、SQLMAP入門 1、判斷是否存在注入 sqlmap.py -u 網址/id=1 id=1不可缺少。當注入點后面的引數大于兩個時。需要加雙引號， sqlmap.py -u "網址/id=1&uid=1" 2、判斷文本中的請求是否存在注入 從文本中加載http請求，SQLMAP可以從一個文本檔案中 ......

    uj5u.com 2020-09-10 02:00:50 more
  • Metasploit 簡單使用教程

    metasploit 簡單使用教程 浩先生， 2020-08-28 16:18:25 分類專欄： kail 網路安全 linux 文章標簽： linux資訊安全 編輯 著作權 metasploit 使用教程 前言 一、Metasploit是什么？ 二、準備作業 三、具體步驟 前言 Msfconsole ......

    uj5u.com 2020-09-10 02:00:53 more
  • 游戲逆向之驅動層與用戶層通訊

    驅動層代碼： #pragma once #include <ntifs.h> #define add_code CTL_CODE(FILE_DEVICE_UNKNOWN,0x800,METHOD_BUFFERED,FILE_ANY_ACCESS) /* 更多游戲逆向視頻www.yxfzedu.com ......

    uj5u.com 2020-09-10 02:00:56 more
  • 北斗電力時鐘（北斗授時服務器）讓網路資料更精準

    北斗電力時鐘（北斗授時服務器）讓網路資料更精準 北斗電力時鐘（北斗授時服務器）讓網路資料更精準 京準電子科技官微——ahjzsz 近幾年，資訊技術的得了快速發展，互聯網在逐漸普及，其在人們生活和生產中都得到了廣泛應用，并且取得了不錯的應用效果。計算機網路資訊在電力系統中的應用，一方面使電力系統的運行 ......

    uj5u.com 2020-09-10 02:01:03 more
  • 【CTF】CTFHub 技能樹 彩蛋 writeup

    ?碎碎念 CTFHub：https://www.ctfhub.com/ 筆者入門CTF時時剛開始刷的是bugku的舊平臺，后來才有了CTFHub。 感覺不論是網頁UI設計，還是題目質量，賽事跟蹤，工具軟體都做得很不錯。 而且因為獨到的金幣制度的確讓人有一種想去刷題賺金幣的感覺。 個人還是非常喜歡這個 ......

    uj5u.com 2020-09-10 02:04:05 more
  • 02windows基礎操作

    我學到了一下幾點 Windows系統目錄結構與滲透的作用 常見Windows的服務詳解 Windows埠詳解 常用的Windows注冊表詳解 hacker DOS命令詳解（net user / type /md /rd/ dir /cd /net use copy、批處理 等） 利用dos命令制作 ......

    uj5u.com 2020-09-10 02:04:18 more
  • 03.Linux基礎操作

    我學到了以下幾點 01Linux系統介紹02系統安裝，密碼啊破解03Linux常用命令04LAMP 01LINUX windows： win03 8 12 16 19 配置不繁瑣 Linux：redhat,centos(紅帽社區版)，Ubuntu server,suse unix:金融機構，證券，銀 ......

    uj5u.com 2020-09-10 02:04:30 more
  • 05HTML

    01HTML介紹 02頭部標簽講解03基礎標簽講解04表單標簽講解 HTML前段語言 js1.了解代碼2.根據代碼 懂得挖掘漏洞 （POST注入/XSS漏洞上傳）3.黑帽seo 白帽seo 客戶網站被黑帽植入劫持代碼如何處理4.熟悉html表單 <html><head><title>TDK標題，描述 ......

    uj5u.com 2020-09-10 02:04:36 more
最新发布
  • 2023年最新微信小程式抓包教程

    01 開門見山 隔一個月發一篇文章，不過分。 首先回顧一下《微信系結手機號資料庫被脫庫事件》，我也是第一時間得知了這個訊息，然后跟蹤了整件事情的經過。下面是這起事件的相關截圖以及近日流出的一萬條資料樣本： 個人認為這件事也沒什么，還不如關注一下之前45億快遞資料查詢渠道疑似在近日復活的訊息。 訊息是 ......

    uj5u.com 2023-04-20 08:48:24 more
  • web3 產品介紹：metamask 錢包 使用最多的瀏覽器插件錢包

    Metamask錢包是一種基于區塊鏈技術的數字貨幣錢包，它允許用戶在安全、便捷的環境下管理自己的加密資產。Metamask錢包是以太坊生態系統中最流行的錢包之一，它具有易于使用、安全性高和功能強大等優點。 本文將詳細介紹Metamask錢包的功能和使用方法。 一、 Metamask錢包的功能 數字資 ......

    uj5u.com 2023-04-20 08:47:46 more
  • vulnhub_Earth

    前言 靶機地址->>>vulnhub_Earth 攻擊機ip：192.168.20.121 靶機ip：192.168.20.122 參考文章 https://www.cnblogs.com/Jing-X/archive/2022/04/03/16097695.html https://www.cnb ......

    uj5u.com 2023-04-20 07:46:20 more
  • 從4k到42k，軟體測驗工程師的漲薪史，給我看哭了

    清明節一過，盲猜大家已經無心上班，在數著日子準備過五一，但一想到銀行卡里的余額……瞬間心情就不美麗了。最近，2023年高校畢業生就業調查顯示，本科畢業月平均起薪為5825元。調查一出，便有很多同學表示自己又被平均了。看著這一資料，不免讓人想到前不久中國青年報的一項調查：近六成大學生認為畢業10年內會 ......

    uj5u.com 2023-04-20 07:44:00 more
  • 最新版本 Stable Diffusion 開源 AI 繪畫工具之中文自動提詞篇

    🎈 標簽生成器 由于輸入正向提示詞 prompt 和反向提示詞 negative prompt 都是使用英文，所以對學習母語的我們非常不友好 使用網址：https://tinygeeker.github.io/p/ai-prompt-generator 這個網址是為了讓大家在使用 AI 繪畫的時候 ......

    uj5u.com 2023-04-20 07:43:36 more
  • 漫談前端自動化測驗演進之路及測驗工具分析

    隨著前端技術的不斷發展和應用程式的日益復雜，前端自動化測驗也在不斷演進。隨著 Web 應用程式變得越來越復雜，自動化測驗的需求也越來越高。如今，自動化測驗已經成為 Web 應用程式開發程序中不可或缺的一部分，它們可以幫助開發人員更快地發現和修復錯誤，提高應用程式的性能和可靠性。 ......

    uj5u.com 2023-04-20 07:43:16 more
  • CANN開發實踐：4個DVPP記憶體問題的典型案例解讀

    摘要：由于DVPP媒體資料處理功能對存放輸入、輸出資料的記憶體有更高的要求（例如，記憶體首地址128位元組對齊），因此需呼叫專用的記憶體申請介面，那么本期就分享幾個關于DVPP記憶體問題的典型案例，并給出原因分析及解決方法。 本文分享自華為云社區《FAQ_DVPP記憶體問題案例》，作者：昇騰CANN。 DVPP ......

    uj5u.com 2023-04-20 07:43:03 more
  • msf學習

    msf學習 以kali自帶的msf為例 一、msf核心模塊與功能 msf模塊都放在/usr/share/metasploit-framework/modules目錄下 1、auxiliary 輔助模塊，輔助滲透（埠掃描、登錄密碼爆破、漏洞驗證等） 2、encoders 編碼器模塊，主要包含各種編碼 ......

    uj5u.com 2023-04-20 07:42:59 more
  • Halcon軟體安裝與界面簡介

    1. 下載Halcon17版本到到本地 2. 雙擊安裝包后 3. 步驟如下 1.2 Halcon軟體安裝 界面分為四大塊 1. Halcon的五個助手 1) 影像采集助手：與相機連接，設定相機引數，采集影像 2) 標定助手：九點標定或是其它的標定，生成標定檔案及內參外參，可以將像素單位轉換為長度單位 ......

    uj5u.com 2023-04-20 07:42:17 more
  • 在MacOS下使用Unity3D開發游戲

    第一次發博客，先發一下我的游戲開發環境吧。 去年2月份買了一臺MacBookPro2021 M1pro(以下簡稱mbp)，這一年來一直在用mbp開發游戲。我大致分享一下我的開發工具以及使用體驗。 1、Unity 官網鏈接： https://unity.cn/releases 我一般使用的Apple ......

    uj5u.com 2023-04-20 07:40:19 more

Copyright © 2010-2020 有解無憂