簡介

Join和coGroup都是flinkSQL中用於連線多個流的運算元，但是有一定的區別，推薦能使用coGroup不要使用Join，因為coGroup更強大。下面讓我們簡單看一下兩個運算元的用法

 CoGrouped原始碼

joinedStream原始碼

對比結論

結論一：.使用時，where裡面要傳入資料流T1要與資料流T2匹配的key，equalTo中傳入T2要和T1匹配的key。

結論二：join和coGroup的最大區別就是apply方法提供的引數型別不一樣，

join的apply

coGroup的apply引數

兩種運算元apply方法中的引數型別不一樣，join中提供的apply方法，引數是T1與T2泛型型別。而coGroup中提供的apply方法，引數是Iterator[T1]與Iterator[2]迭代器，基於這2種方式，如果我們的邏輯不僅僅是對一條record做處理，而是要與上一record或更復雜的判斷與比較，甚至是對結果排序，那麼join中的apply方法顯得比較困難，所以推薦使用coGroup

我們想要在Flink中實現實時的流計算，就可以通過joinedStream或coGroupedStream來實現。但是在join之後實施更復雜的運算，例如判斷、迭代等，僅僅通過SQL實現，恐怕會很麻煩，只能通過PL/SQL塊來實現，而Flink提供了apply方法，使用者可以自己編寫複雜的函式來實現。

案例 雙流jion

package flinkSQL
import java.text.SimpleDateFormat
import org.apache.flink.api.common.functions.CoGroupFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala._

/**
  * Created by  ${WangZhiHua} on 2018/11/13
  */
object JoinDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //獲取介面傳送的資料
    val dataStream1 = env.readTextFile("C:/flink_data/scoket1.txt")
    val dataStream2 = env.readTextFile("C:/flink_data/scoket2.txt")
    val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
   //使用樣例類StockTransaction封裝獲取的資料
    val dataStreamMap1 = dataStream1.map(f => {
      val tokens1 = f.split(",")
      StockTransaction(tokens1(0), tokens1(1), tokens1(2).toDouble)
    })
      .assignAscendingTimestamps(f => format.parse(f.tx_time).getTime)
    //使用樣例類StockSnapshot封裝獲取的資料
    val dataStreamMap2 = dataStream2.map(f => {
      val tokens2 = f.split(",")
      StockSnapshot(tokens2(0), tokens2(1), tokens2(2).toDouble)
    })
      .assignAscendingTimestamps(f => format.parse(f.md_time).getTime)

    /**
      * 進行雙流join
      * 限定範圍是：3秒鐘的Event time時間視窗
      */

    val joinStream = dataStreamMap1.coGroup(dataStreamMap2)
      .where(_.tx_code)
      .equalTo(_.md_code)
      .window(TumblingEventTimeWindows.of(Time.seconds(3)))

      val innerJoinStream = joinStream.apply(new InnerJoinFunction)
     innerJoinStream.name("innerJoin").print()
    print("===================== end =========================")
    env.execute("join demo")
  }
}
//定義樣例類封裝接收的資料
case class StockTransaction(tx_time:String, tx_code:String,tx_value:Double)
case class StockSnapshot(md_time:String, md_code:String,md_value:Double)
//定義一個內連線函式，繼承CoCroup
class InnerJoinFunction extends CoGroupFunction[StockTransaction,StockSnapshot,(String,String,String,Double,Double,String)]{
  override def coGroup(T1: java.lang.Iterable[StockTransaction], T2: java.lang.Iterable[StockSnapshot], out: Collector[(String, String, String, Double, Double,String)]): Unit = {
    /**
      * 將Java中的Iterable物件轉換為Scala的Iterable
      * scala的集合操作效率高，簡潔
      */
    import scala.collection.JavaConverters._
    val scalaT1 = T1.asScala.toList
    val scalaT2 = T2.asScala.toList

    /**
      * Inner Join要比較的是同一個key下，同一個時間視窗內的資料
      */
    if(scalaT1.nonEmpty && scalaT2.nonEmpty){
      for(transaction <- scalaT1){
        for(snapshot <- scalaT2){
          out.collect(transaction.tx_code,transaction.tx_time, snapshot.md_time,transaction.tx_value,snapshot.md_value,"Inner Join Test")
        }
      }
    }
  }
}
class LeftJoinFunction extends CoGroupFunction[StockTransaction,StockSnapshot,(String,String,String,Double,Double,String)] {
  override def coGroup(T1: java.lang.Iterable[StockTransaction], T2: java.lang.Iterable[StockSnapshot], out: Collector[(String, String, String, Double,Double,String)]): Unit = {
    /**
      * 將Java中的Iterable物件轉換為Scala的Iterable
      * scala的集合操作效率高，簡潔
      */
    import scala.collection.JavaConverters._
    val scalaT1 = T1.asScala.toList
    val scalaT2 = T2.asScala.toList

    /**
      * Left Join要比較的是同一個key下，同一個時間視窗內的資料
      */
    if(scalaT1.nonEmpty && scalaT2.isEmpty){
      for(transaction <- scalaT1){
        out.collect(transaction.tx_code,transaction.tx_time, "",transaction.tx_value,0,"Left Join Test")
      }
    }
  }
}
class RightJoinFunction extends CoGroupFunction[StockTransaction,StockSnapshot,(String,String,String,Double,Double,String)] {
  override def coGroup(T1: java.lang.Iterable[StockTransaction], T2: java.lang.Iterable[StockSnapshot], out: Collector[(String, String, String, Double,Double,String)]): Unit = {
    /**
      * 將Java中的Iterable物件轉換為Scala的Iterable
      * scala的集合操作效率高，簡潔
      */
    import scala.collection.JavaConverters._
    val scalaT1 = T1.asScala.toList
    val scalaT2 = T2.asScala.toList

    /**
      * Right Join要比較的是同一個key下，同一個時間視窗內的資料
      */
    if(scalaT1.isEmpty && scalaT2.nonEmpty){
      for(snapshot <- scalaT2){
        out.collect(snapshot.md_code, "",snapshot.md_time,0,snapshot.md_value,"Right Join Test")
      }
    }
  }
}