傳送區(qū)

[Scala] Flink項(xiàng)目實(shí)例系列（零）
[Scala] Flink項(xiàng)目實(shí)時熱門商品統(tǒng)計（一）
[Scala] Flink項(xiàng)目實(shí)時流量統(tǒng)計（二）
[Scala] Flink項(xiàng)目惡意登錄監(jiān)控（三）
[Scala] Flink項(xiàng)目訂單支付失效監(jiān)控（四）
[Scala] Flink項(xiàng)目訂單支付實(shí)時對賬（五）
[Scala] Flink項(xiàng)目小彩蛋（六）
本項(xiàng)目的代碼及文件見這這這，友情碼是：3n9z。

Join

Join官方傳送

戳我

Tumbling Window Join

圖好話少

Sliding Window Join

Session Window Join

Interval Join

示例代碼

數(shù)據(jù)源結(jié)構(gòu)

orderIdeventTypetxIdtimestamp

34729	pay	sd76f87d6	1558430844

txIdpayChanneleventTime

ewr342as4	wechat	1558430845

import org.apache.flink.streaming.api.TimeCharacteristic import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time import org.apache.flink.util.Collectorobject TxMatchByJoin {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 讀取訂單事件流val resource = getClass.getResource("/OrderLog.csv")val orderEventStream = env.readTextFile(resource.getPath)// val orderEventStream = env.socketTextStream("localhost", 7777).map(data => {val dataArray = data.split(",")OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)}).filter(_.txId != "").assignAscendingTimestamps(_.eventTime * 1000L).keyBy(_.txId)// 讀取支付到賬事件流val receiptResource = getClass.getResource("/ReceiptLog.csv")val receiptEventStream = env.readTextFile(receiptResource.getPath) // val receiptEventStream = env.socketTextStream("localhost", 8888).map(data => {val dataArray = data.split(",")ReceiptEvent(dataArray(0).trim, dataArray(1).trim, dataArray(2).toLong)}).assignAscendingTimestamps(_.eventTime * 1000L).keyBy(_.txId)// intervalJoin的概念見鏈接// https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/operators/joining.html#interval-joinval processedStream = orderEventStream.intervalJoin(receiptEventStream).between(Time.seconds(-5), Time.seconds(5)).process(new TxPayMatchByJoin())processedStream.print()env.execute("tx pay match by join job")} }class TxPayMatchByJoin() extends ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent,ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {out.collect((left, right))} } ?

總結(jié)

以上是生活随笔為你收集整理的[Scala] Flink项目小彩蛋（六）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。