文章目錄
- 一.簡介
- 二.視窗Join
- 2.1 翻滾視窗（Tumbling Window Join）
- 2.2 滑動視窗Join（Sliding Window Join）
- 2.3 會話視窗Join（Session Window Join）
- 2.4.小結
- 三.間隔Join
- 四.示例
- 4.1 間隔Join
- 4.2 視窗Join
一.簡介
Flink DataStream API中內置有兩個可以根據實際條件對資料流進行Join算子：基于間隔的Join和基于視窗的Join，
語意注意事項
- 創建兩個流元素的成對組合的行為類似內連接，如果來自一個流的元素與另一個流沒有相對應要連接的元素，則不會發出該元素，
- 結合在一起的那些元素將其時間戳設定為位于各自視窗中最大時間戳，例如：以[5,10]為邊界的視窗將產生連接的元素的時間戳為9，
二.視窗Join
2.1 翻滾視窗（Tumbling Window Join）
執行滾動視窗連接（Tumbling Window Join）時，具有公共Key和公共Tumbling Window的所有元素都以成對組合形式進行連接，并傳遞給JoinFunction或FlatJoinFunction，因為這就像一個內連接，在滾動視窗中沒有來自另一個流的元素的流的元素不會被輸出，
如圖所示，我們定義了一個大小為2毫秒的滾動視窗，其結果為[0,1]，[2,3]， …，該影像顯示了每個視窗中所有元素的成對組合，這些元素將傳遞給JoinFunction，注意，在翻滾視窗[6,7]中沒有發出任何內容，因為在綠色流中沒有元素與橙色元素⑥、⑦連接，
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
...
val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...
orangeStream.join(greenStream)
.where(elem => /* select key */)
.equalTo(elem => /* select key */)
.window(TumblingEventTimeWindows.of(Time.milliseconds(2)))
.apply { (e1, e2) => e1 + "," + e2 }
2.2 滑動視窗Join（Sliding Window Join）
在執行滑動視窗連接（Sliding Window Join）時，具有公共Key和公共滑動視窗（Sliding Window ）的所有元素都作為成對組合進行連接，并傳遞給JoinFunction或FlatJoinFunction，當前滑動視窗中沒有來自另一個流的元素的流的元素不會被發出，
注意，有些元素可能會在一個滑動視窗中連接，但不會在另一個視窗中連接!
在本例中，我們使用的滑動視窗大小為2毫秒，滑動1毫秒，滑動視窗結果[1,0],[0,1],[1,2],[2、3],… x軸以下是每個滑動視窗的Join結果將被傳遞給JoinFunction的元素，在這里你還可以看到橙②與綠色③視窗Join(2、3),但不與任何視窗Join[1,2]，
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
...
val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...
orangeStream.join(greenStream)
.where(elem => /* select key */)
.equalTo(elem => /* select key */)
.window(SlidingEventTimeWindows.of(Time.milliseconds(2) /* size */, Time.milliseconds(1) /* slide */))
.apply { (e1, e2) => e1 + "," + e2 }
2.3 會話視窗Join（Session Window Join）
在執行會話視窗連接時，具有相同鍵的所有元素(當“組合”時滿足會話條件)都以成對的組合進行連接，并傳遞給JoinFunction或FlatJoinFunction，再次執行內部連接，因此如果會話視窗只包含來自一個流的元素，則不會發出任何輸出，
在這里，定義一個會話視窗連接，其中每個會話被至少1ms的間隔所分割，有三個會話，在前兩個會話中，來自兩個流的連接元素被傳遞給JoinFunction，在第三次會話中綠色流沒有元素,所以⑧⑨不會Join，
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
...
val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...
orangeStream.join(greenStream)
.where(elem => /* select key */)
.equalTo(elem => /* select key */)
.window(EventTimeSessionWindows.withGap(Time.milliseconds(1)))
.apply { (e1, e2) => e1 + "," + e2 }
2.4.小結
除了對視窗中兩條流進行Join，你還可以對它們進行Cogroup，只需將算子定義開始位置的Join()改為coGroup()即可，Join和Cogroup的總體邏輯相同，
二者區別：Join會為兩側輸入中每個事件對呼叫JoinFunction;而Cogroup中CoGroupFunction會以兩個輸入的元素遍歷器為引數，只在每個視窗中被呼叫一次，
三.間隔Join
interval join用一個公共Key連接兩個流的元素(將它們稱為A & B)，其中流B的元素的時間戳具有相對于流A中的元素的時間戳， 這也可以更正式地表示為b.timestamp ∈ [a.timestamp + lowerBound; a.timestamp + upperBound] or a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound
其中a和b是A和B中共享一個公鑰的元素，下界和上界都可以是負的或正的，只要下界小于或等于上界，interval連接目前只執行內部連接，
當將一對元素傳遞給ProcessJoinFunction時，它們將給兩個元素分配更大的時間戳(可以通過ProcessJoinFunction.Context訪問)，
注意：間隔連接目前只支持事件時間，
在上面的示例中，我們將“橙色”和“綠色”兩個流連接起來，它們的下界為-2毫秒，上界為+1毫秒，默認情況下，這些是包含邊界的，但是可以通過.lowerboundexclusive()和. upperboundexclusive()進行設定，
再用更正式的符號來表示angeElem.ts + lowerBound <= greenElem.ts <= orangeElem.ts + upperBound 如三角形所示，
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
...
val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...
orangeStream
.keyBy(elem => /* select key */)
.intervalJoin(greenStream.keyBy(elem => /* select key */))
.between(Time.milliseconds(-2), Time.milliseconds(1))
.process(new ProcessJoinFunction[Integer, Integer, String] {
override def processElement(left: Integer, right: Integer, ctx: ProcessJoinFunction[Integer, Integer, String]#Context, out: Collector[String]): Unit = {
out.collect(left + "," + right);
}
});
});
四.示例
4.1 間隔Join
package com.lm.flink.datastream.join
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
/**
* @Classname IntervalJoin
* @Description TODO
* @Date 2020/10/27 20:32
* @Created by limeng
* 區間關聯當前僅支持EventTime
* Interval JOIN 相對于UnBounded的雙流JOIN來說是Bounded JOIN，就是每條流的每一條資料會與另一條流上的不同時間區域的資料進行JOIN，
*/
object IntervalJoin {
def main(args: Array[String]): Unit = {
//設定至少一次或僅此一次語意
val env = StreamExecutionEnvironment.getExecutionEnvironment
//設定至少一次或僅此一次語意
env.enableCheckpointing(20000,CheckpointingMode.EXACTLY_ONCE)
//設定
env.getCheckpointConfig
.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
//設定重啟策略
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5,50000))
env.setParallelism(1)
val dataStream1 = env.socketTextStream("localhost",9999)
val dataStream2 = env.socketTextStream("localhost",9998)
import org.apache.flink.api.scala._
val dataStreamMap1 = dataStream1.map(f=>{
val tokens = f.split(",")
StockTransaction(tokens(0),tokens(1),tokens(2).toDouble)
}).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StockTransaction]{
var currentTimestamp = 0L
val maxOutOfOrderness = 1000L
override def getCurrentWatermark: Watermark = {
val tmpTimestamp = currentTimestamp - maxOutOfOrderness
println(s"wall clock is ${System.currentTimeMillis()} new watermark ${tmpTimestamp}")
new Watermark(tmpTimestamp)
}
override def extractTimestamp(element: StockTransaction, previousElementTimestamp: Long): Long = {
val timestamp = element.txTime.toLong
currentTimestamp = Math.max(timestamp,currentTimestamp)
println(s"get timestamp is $timestamp currentMaxTimestamp $currentTimestamp")
currentTimestamp
}
})
val dataStreamMap2 = dataStream2.map(f=>{
val tokens = f.split(",")
StockSnapshot(tokens(0),tokens(1),tokens(2).toDouble)
}).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StockSnapshot]{
var currentTimestamp = 0L
val maxOutOfOrderness = 1000L
override def getCurrentWatermark: Watermark = {
val tmpTimestamp = currentTimestamp - maxOutOfOrderness
println(s"wall clock is ${System.currentTimeMillis()} new watermark ${tmpTimestamp}")
new Watermark(tmpTimestamp)
}
override def extractTimestamp(element: StockSnapshot, previousElementTimestamp: Long): Long = {
val timestamp = element.mdTime.toLong
currentTimestamp = Math.max(timestamp,currentTimestamp)
println(s"get timestamp is $timestamp currentMaxTimestamp $currentTimestamp")
currentTimestamp
}
})
dataStreamMap1.print("dataStreamMap1")
dataStreamMap2.print("dataStreamMap2")
dataStreamMap1.keyBy(_.txCode)
.intervalJoin(dataStreamMap2.keyBy(_.mdCode))
.between(Time.minutes(-10),Time.seconds(0))
.process(new ProcessJoinFunction[StockTransaction,StockSnapshot,String] {
override def processElement(left: StockTransaction, right: StockSnapshot, ctx: ProcessJoinFunction[StockTransaction, StockSnapshot, String]#Context, out: Collector[String]): Unit = {
out.collect(left.toString +" =Interval Join=> "+right.toString)
}
}).print()
env.execute("IntervalJoin")
}
case class StockTransaction(txTime:String,txCode:String,txValue:Double) extends Serializable{
override def toString: String = txTime +"#"+txCode+"#"+txValue
}
case class StockSnapshot(mdTime:String,mdCode:String,mdValue:Double) extends Serializable {
override def toString: String = mdTime +"#"+mdCode+"#"+mdValue
}
}
結果
get timestamp is 1603708942 currentMaxTimestamp 1603708942
dataStreamMap1> 1603708942#000001#10.4
get timestamp is 1603708942 currentMaxTimestamp 1603708942
dataStreamMap2> 1603708942#000001#10.4
1603708942#000001#10.4 =Interval Join=> 1603708942#000001#10.4
4.2 視窗Join
package com.lm.flink.datastream.join
import java.lang
import org.apache.flink.api.common.functions.CoGroupFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
/**
* @Classname InnerLeftRightJoinTest
* @Description TODO
* @Date 2020/10/26 17:22
* @Created by limeng
* window join
*/
object InnerLeftRightJoinTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//每9秒發出一個watermark
env.setParallelism(1)
env.getConfig.setAutoWatermarkInterval(9000)
val dataStream1 = env.socketTextStream("localhost", 9999)
val dataStream2 = env.socketTextStream("localhost", 9998)
/**
* operator操作
* 資料格式：
* tx: 2020/10/26 18:42:22,000002,10.2
* md: 2020/10/26 18:42:22,000002,10.2
*
* 這里由于是測驗，固水位線采用升序（即資料的Event Time 本身是升序輸入）
*/
import org.apache.flink.api.scala._
val dataStreamMap1 = dataStream1
.map(f => {
val tokens = f.split(",")
StockTransaction(tokens(0), tokens(1), tokens(2).toDouble)
}).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StockTransaction] {
var currentTimestamp = 0L
val maxOutOfOrderness = 1000L
override def getCurrentWatermark: Watermark = {
val tmpTimestamp = currentTimestamp - maxOutOfOrderness
println(s"wall clock is ${System.currentTimeMillis()} new watermark ${tmpTimestamp}")
new Watermark(tmpTimestamp)
}
override def extractTimestamp(element: StockTransaction, previousElementTimestamp: Long): Long = {
val timestamp = element.txTime.toLong
currentTimestamp = Math.max(timestamp, currentTimestamp)
println(s"get timestamp is $timestamp currentMaxTimestamp $currentTimestamp")
currentTimestamp
}
})
val dataStreamMap2 = dataStream2
.map(f => {
val tokens = f.split(",")
StockSnapshot(tokens(0), tokens(1), tokens(2).toDouble)
}).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StockSnapshot] {
var currentTimestamp = 0L
val maxOutOfOrderness = 1000L
override def getCurrentWatermark: Watermark = {
val tmpTimestamp = currentTimestamp - maxOutOfOrderness
println(s"wall clock is ${System.currentTimeMillis()} new watermark ${tmpTimestamp}")
new Watermark(tmpTimestamp)
}
override def extractTimestamp(element: StockSnapshot, previousElementTimestamp: Long): Long = {
val timestamp = element.mdTime.toLong
currentTimestamp = Math.max(timestamp, currentTimestamp)
println(s"get timestamp is $timestamp currentMaxTimestamp $currentTimestamp")
currentTimestamp
}
})
dataStreamMap1.print("dataStreamMap1")
dataStreamMap2.print("dataStreamMap2")
/**
* Join操作
* 限定范圍是3秒鐘的Event Time視窗
*/
val joinedStream = dataStreamMap1.coGroup(dataStreamMap2)
.where(_.txCode)
.equalTo(_.mdCode)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
val innerJoinedStream = joinedStream.apply(new InnerJoinFunction)
val leftJoinedStream = joinedStream.apply(new LeftJoinFunction)
val rightJoinedStream = joinedStream.apply(new RightJoinFunction)
innerJoinedStream.name("InnerJoinedStream").print()
leftJoinedStream.name("LeftJoinedStream").print()
rightJoinedStream.name("RightJoinedStream").print()
env.execute("InnerLeftRightJoinTest")
}
class InnerJoinFunction extends CoGroupFunction[StockTransaction, StockSnapshot, (String, String, String, Double, Double, String)] {
override def coGroup(first: lang.Iterable[StockTransaction], second: lang.Iterable[StockSnapshot], out: Collector[(String, String, String, Double, Double, String)]): Unit = {
import scala.collection.JavaConverters._
val scalaT1 = first.asScala.toList
val scalaT2 = second.asScala.toList
println(scalaT1.size)
println(scalaT2.size)
/**
* Inner join 要比較的是同一個key下，同一個時間視窗內
*/
if (scalaT1.nonEmpty && scalaT2.nonEmpty) {
for (transaction <- scalaT1) {
for (snapshot <- scalaT2) {
out.collect(transaction.txCode, transaction.txTime, snapshot.mdTime, transaction.txValue, snapshot.mdValue, "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.txCode, transaction.txTime, "", transaction.txValue, 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.mdCode, "", snapshot.mdTime, 0, snapshot.mdValue, "Right Join Test")
}
}
}
}
case class StockTransaction(txTime: String, txCode: String, txValue: Double)
case class StockSnapshot(mdTime: String, mdCode: String, mdValue: Double)
}
參考
https://www.jianshu.com/p/ba19e4d1d802
公眾號
名稱：大資料計算
微信號：bigdata_limeng
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/197939.html
標籤：其他