快速入門Flink(8)——Flink中的流式處理Transformation操作

        上篇部落格給大家講解了DataSource與DataSink本篇文章準備給大家講解下Stream中的最長用的幾種Transformation操作（收藏，收藏，收藏重要事情說三遍）。

一、KeyBy

邏輯上將一個流分成不相交的分割區，每個分割區包含相同鍵的元素。在內部，這是通過散 列分割區來實現的

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

/**
 * @author
 * @date 2020/9/23 21:50
 * @version 1.0
 */
object StreamKeyBy {
  def main(args: Array[String]): Unit = {
    //1.構建流處理執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用socket構建資料來源
    val socketDataSource = env.socketTextStream("node01", 9999)
    //3.處理資料
    val keyBy = socketDataSource.flatMap(_.split(" ")).map((_, 1)).keyBy(0)
    //4.輸出
    keyBy.print("StreamKeyBy")
    //5.任務執行
    env.execute("StreamKeyBy")
  }
}

二、Connect

用來將兩個 dataStream
組裝成一個 ConnectedStreams 而且這個 connectedStream 的組成結構就是保留原有的 dataStream 的結構體；這樣我們 就可以把不同的資料組裝成同一個結構

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

/**
 * @author
 * @date 2020/9/23 22:03
 * @version 1.0
 */
object StreamConnect {
  def main(args: Array[String]): Unit = {
    //1.構建批次處理執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.構建2個資料流
    val source1 = env.addSource(new MyNoParallelSource).setParallelism(1)
    val source2 = env.addSource(new MyNoParallelSource).setParallelism(1)
    //3.使用合併流
    val connectStream = source1.connect(source2)
    val result = connectStream.map(function1 => {
      "function1" + function1
    }, function2 => {
      "function2" + function2
    })
    //4.輸出
    result.print()
    //5.任務啟動
    env.execute("StreamConnect")
  }

  class MyNoParallelSource() extends SourceFunction[Long] {
    var count = 1L
    var isRunning = true

    override def run(sourceContext: SourceFunction.SourceContext[Long]): Unit = {
      while (isRunning) {
        sourceContext.collect(count)
        count += 1
        Thread.sleep(1000)
        if (count > 5) {
          cancel()
        }
      }
    }

    override def cancel(): Unit = {
      isRunning = false
    }
  }
}

三、Split 和 select

Split 就是將一個 DataStream 分成兩個或者多個 DataStream Select 就是獲取分流後對應的資料
需求： 給出資料 1, 2, 3, 4, 5, 6, 7
請使用 split 和 select 把資料中的奇偶數分開，並列印出奇數

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * @author
 * @date 2020/9/23 22:14
 * @version 1.0
 */
object StreamSplit {
  def main(args: Array[String]): Unit = {
    //1.構建流處理執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val source = env.generateSequence(1, 10)
    //3.使用split將資料進行切分
    val splitStream = source.split(x => {
      (x % 2) match {
        case 0 => List("偶數")
        case 1 => List("奇數")
      }
    })
    //4.獲取奇數並列印
    val result = splitStream.select("奇數")
    result.print()
    //5.任務執行
    env.execute("StreamSplit")
  }
}