快速入門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")
  }
}