connect 算子

Connect算子主要是为了合并两种不同数据类型的数据集，合并后会保留原来数据集的数据类型。例如：dataStream1数据集为(String, Int)元祖类型，dataStream2数据集为Int类型，通过connect连接算子将两个不同数据类型的流结合在一起，形成格式为ConnectedStreams的数据集，其内部数据为[(String, Int), Int]的混合数据类型，保留了两个原始数据集的数据类型。

xxxxxxxxxx
package com.itbaizhan.flink.scala.transformation

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

object ConnectDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //添加隐式转
    import org.apache.flink.streaming.api.scala._
    //创建两个不同类型的数据集
    val dataStream1: DataStream[(String, Int)] = env.fromElements(("a", 3), ("d", 4), ("c", 2))
    val dataStream2: DataStream[Int] = env.fromElements(1, 2, 4)
    //连接两个数据集
    //ConnectedStreams类型的数据集不能直接进行类似print()操作
    val resultCS: ConnectedStreams[(String, Int), Int] = dataStream1.connect(dataStream2)
    //需要将之转化为DataStream类型数据集，在Flink中ConnectedStreams提供了map()和flatMap()
    val resultDS: DataStream[(Any, Int)] = resultCS.map(
      data1 => {
        (data1._1, data1._2)
      },
      data2 => {
        (data2, 0)
      }
    )
    resultDS.print()
    env.execute("ConnectDemo")
  }
}

输出：

(1,0) (2,0) (a,3) (4,0) (d,4) (c,2)

注意Union和Connect的区别：

• union之前两个流的类型必须是一样，Connect可以不一样，在之后的map中再去调整成为一样的。

• connect只能操作两个流，union可以操作多个。