Flink算子使用方法及实例演示：union和connect

向右看齐 2022-11-15 14:07 61阅读 0赞

Flink的Transformation转换主要包括四种：单数据流基本转换、基于Key的分组转换、多数据流转换和数据重分布转换。读者可以使用Flink Scala Shell或者Intellij Idea来进行练习：

*  [Flink Scala Shell使用教程][Flink Scala Shell]
 *  [Intellij Idea开发环境搭建教程][Intellij Idea]
 *  [Flink单数据流基本转换：map、filter、flatMap][Flink_map_filter_flatMap]
 *  [Flink基于Key的分组转换：keyBy、reduce和aggregations][Flink_Key_keyBy_reduce_aggregations]

很多情况下，我们需要对多个数据流进行整合处理，Flink为我们提供了多流转换算子，本文主要介绍多流转换。

## union ##

在`DataStream`上使用`union`算子可以合并多个同类型的数据流，并生成同类型的数据流，即可以将多个`DataStream[T]`合并为一个新的`DataStream[T]`。数据将按照先进先出（First In First Out）的模式合并，且不去重。下图`union`对白色和深色两个数据流进行合并，生成一个数据流。

![f93ac3d778ef789621f04ffe062b30fc.png][]

假设股票价格数据流来自不同的交易所，我们将其合并成一个数据流：

val shenzhenStockStream: DataStream[StockPrice] = ...
    val hongkongStockStream: DataStream[StockPrice] = ...
    val shanghaiStockStream: DataStream[StockPrice] = ...
    val unionStockStream: DataStream[StockPrice] = shenzhenStockStream.union(hongkongStockStream, shanghaiStockStream)

## connect ##

`union`虽然可以合并多个数据流，但有一个限制，即多个数据流的数据类型必须相同。`connect`提供了和`union`类似的功能，用来连接两个数据流，它与`union`的区别在于：

1.  `connect`只能连接两个数据流，`union`可以连接多个数据流。
2.  `connect`所连接的两个数据流的数据类型可以不一致，`union`所连接的两个数据流的数据类型必须一致。
3.  两个`DataStream`经过`connect`之后被转化为`ConnectedStreams`，`ConnectedStreams`会对两个流的数据应用不同的处理方法，且双流之间可以共享状态。

`connect`经常被应用在对一个数据流使用另外一个流进行控制处理的场景上，如下图所示。控制流可以是阈值、规则、机器学习模型或其他参数。

![44b2ac790879b78a5c3f64c9261dfdfd.png][]

对于`ConnectedStreams`，我们需要重写`CoMapFunction`或`CoFlatMapFunction`。这两个接口都提供了三个泛型，这三个泛型分别对应第一个输入流的数据类型、第二个输入流的数据类型和输出流的数据类型。在重写函数时，对于`CoMapFunction`，`map1`处理第一个流的数据，`map2`处理第二个流的数据；对于`CoFlatMapFunction`，`flatMap1`处理第一个流的数据，`flatMap2`处理第二个流的数据。Flink并不能保证两个函数调用顺序，两个函数的调用依赖于两个数据流数据的流入先后顺序，即第一个数据流有数据到达时，`map1`或`flatMap1`会被调用，第二个数据流有数据到达时，`map2`或`flatMap2`会被调用。下面的代码对一个整数流和一个字符串流进行了`connect`操作。

val intStream: DataStream[Int] = senv.fromElements(1, 0, 9, 2, 3, 6)
    val stringStream: DataStream[String] = senv.fromElements("LOW", "HIGH", "LOW", "LOW")
    
    val connectedStream: ConnectedStreams[Int, String] = intStream.connect(stringStream)
    
    // CoMapFunction三个泛型分别对应第一个流的输入、第二个流的输入，map之后的输出
    class MyCoMapFunction extends CoMapFunction[Int, String, String] {
    
      override def map1(input1: Int): String = input1.toString
    
      override def map2(input2: String): String = input2
    }
    
    val mapResult = connectedStream.map(new MyCoMapFunction)

我们知道，如果不对`DataStream`按照Key进行分组，数据是随机分配在各个TaskSlot上的，而绝大多数情况我们是要对某个Key进行分析和处理，Flink允许我们将`connect`和`keyBy`或`broadcast`结合起来使用。例如，我们将之前的股票价格数据流与一个媒体评价数据流结合起来，按照股票代号进行分组。

// 先将两个流connect，再进行keyBy
    val keyByConnect1: ConnectedStreams[StockPrice, Media] = stockPriceRawStream
      .connect(mediaStatusStream)
      .keyBy(0,0)
    
    // 先keyBy再connect
    val keyByConnect2: ConnectedStreams[StockPrice, Media] = stockPriceRawStream.keyBy(0)
        .connect(mediaStatusStream.keyBy(0))

无论先`keyBy`还是先`connect`，我们都可以将含有相同Key的数据转发到下游同一个算子实例上。这种操作有点像SQL中的join操作。Flink也提供了`join`算子，`join`主要在时间窗口维度上，`connect`相比而言更广义一些，关于`join`的介绍将在后续文章中介绍。

下面的代码展示了如何将股票价格和媒体正负面评价结合起来，当媒体评价为正且股票价格大于阈值时，输出一个正面信号。完整代码在我的github上：[https://github.com/luweizheng/flink-tutorials][https_github.com_luweizheng_flink-tutorials]

package com.flink.tutorials.demos.stock
    
    import java.util.Calendar
    
    import com.flink.tutorials.demos.stock.StockPriceDemo.{StockPrice, StockPriceSource, StockPriceTimeAssigner}
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction
    import org.apache.flink.streaming.api.functions.source.RichSourceFunction
    import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    import scala.util.Random
    
    object StockMediaConnectedDemo {
    
      def main(args: Array[String]) {
    
        // 设置执行环境
        val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration())
    
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        // 每5秒生成一个Watermark
        env.getConfig.setAutoWatermarkInterval(5000L)
    
        // 股票价格数据流
        val stockPriceRawStream: DataStream[StockPrice] = env
          // 该数据流由StockPriceSource类随机生成
          .addSource(new StockPriceSource)
          // 设置 Timestamp 和 Watermark
          .assignTimestampsAndWatermarks(new StockPriceTimeAssigner)
    
        val mediaStatusStream: DataStream[Media] = env
          .addSource(new MediaSource)
    
        // 先将两个流connect，再进行keyBy
        val keyByConnect1: ConnectedStreams[StockPrice, Media] = stockPriceRawStream
          .connect(mediaStatusStream)
          .keyBy(0,0)
    
        // 先keyBy再connect
        val keyByConnect2: ConnectedStreams[StockPrice, Media] = stockPriceRawStream.keyBy(0)
          .connect(mediaStatusStream.keyBy(0))
    
        val alert1 = keyByConnect1.flatMap(new AlertFlatMap).print()
    
        val alerts2 = keyByConnect2.flatMap(new AlertFlatMap).print()
    
        // 执行程序
        env.execute("connect stock price with media status")
      }
    
      /** 媒体评价
        *
        * symbol 股票代号
        * timestamp 时间戳
        * status 评价 正面/一般/负面
        */
      case class Media(symbol: String, timestamp: Long, status: String)
    
      class MediaSource extends RichSourceFunction[Media]{
    
        var isRunning: Boolean = true
    
        val rand = new Random()
        var stockId = 0
    
        override def run(srcCtx: SourceContext[Media]): Unit = {
    
          while (isRunning) {
            // 每次从列表中随机选择一只股票
            stockId = rand.nextInt(5)
    
            var status: String = "NORMAL"
            if (rand.nextGaussian() > 0.9) {
              status = "POSITIVE"
            } else if (rand.nextGaussian() < 0.05) {
              status = "NEGATIVE"
            }
    
            val curTime = Calendar.getInstance.getTimeInMillis
    
            srcCtx.collect(Media(stockId.toString, curTime, status))
    
            Thread.sleep(rand.nextInt(100))
          }
        }
    
        override def cancel(): Unit = {
          isRunning = false
        }
      }
    
      case class Alert(symbol: String, timestamp: Long, alert: String)
    
      class AlertFlatMap extends RichCoFlatMapFunction[StockPrice, Media, Alert] {
    
        var priceMaxThreshold: List[Double] = List(101.0d, 201.0d, 301.0d, 401.0d, 501.0d)
    
        var mediaLevel: String = "NORMAL"
    
        override def flatMap1(stock: StockPrice, collector: Collector[Alert]) : Unit = {
          val stockId = stock.symbol.toInt
          if ("POSITIVE".equals(mediaLevel) && stock.price > priceMaxThreshold(stockId)) {
            collector.collect(Alert(stock.symbol, stock.timestamp, "POSITIVE"))
          }
        }
    
        override def flatMap2(media: Media, collector: Collector[Alert]): Unit = {
          mediaLevel = media.status
        }
      }
    
    }

[Flink Scala Shell]: https://link.zhihu.com/?target=https%3A//mp.weixin.qq.com/s/UOoUo5QWUILKtoTG4gC4hQ
[Intellij Idea]: https://link.zhihu.com/?target=https%3A//mp.weixin.qq.com/s/S3anHu_d1wAwf8_tIgnDfw
[Flink_map_filter_flatMap]: https://link.zhihu.com/?target=https%3A//mp.weixin.qq.com/s/z8L6QU1ZWW1-O2cn8ixcmg
[Flink_Key_keyBy_reduce_aggregations]: https://link.zhihu.com/?target=https%3A//mp.weixin.qq.com/s/2vcKteQIyj31sVrSg1R_2Q
[f93ac3d778ef789621f04ffe062b30fc.png]: /images/20221022/dd4b5877670f4ea1863712220a0f4548.png
[44b2ac790879b78a5c3f64c9261dfdfd.png]: /images/20221022/15ea67bab62b44dfb53a23576f74ce85.png
[https_github.com_luweizheng_flink-tutorials]: https://link.zhihu.com/?target=https%3A//github.com/luweizheng/flink-tutorials