Flink 中極其重要的 Time 與 Window 詳細解析(深度好文，建議收藏)

時間 2021-01-26

標籤 java node python web sql shell apache json bootstrap vim 欄目 Windows 简体版

原文原文鏈接

前言

Flink 是流式的、實時的計算引擎java

上面一句話就有兩個概念，一個是流式，一個是實時。node

流式：就是數據源源不斷的流進來，也就是數據沒有邊界，可是咱們計算的時候必須在一個有邊界的範圍內進行，因此這裏面就有一個問題，邊界怎麼肯定？無非就兩種方式，根據時間段或者數據量進行肯定，根據時間段就是每隔多長時間就劃分一個邊界，根據數據量就是每來多少條數據劃分一個邊界，Flink 中就是這麼劃分邊界的，本文會詳細講解。python

實時：就是數據發送過來以後立馬就進行相關的計算，而後將結果輸出。這裏的計算有兩種：web

一種是隻有邊界內的數據進行計算，這種好理解，好比統計每一個用戶最近五分鐘內瀏覽的新聞數量，就能夠取最近五分鐘內的全部數據，而後根據每一個用戶分組，統計新聞的總數。sql
另外一種是邊界內數據與外部數據進行關聯計算，好比：統計最近五分鐘內瀏覽新聞的用戶都是來自哪些地區，這種就須要將五分鐘內瀏覽新聞的用戶信息與 hive 中的地區維表進行關聯，而後在進行相關計算。shell

本篇文章所講的 Flink 的內容就是圍繞以上概念進行詳細剖析的！apache

Time與Window

Time

在Flink中，若是以時間段劃分邊界的話，那麼時間就是一個極其重要的字段。json

Flink中的時間有三種類型，以下圖所示：bootstrap

Event Time：是事件建立的時間。它一般由事件中的時間戳描述，例如採集的日誌數據中，每一條日誌都會記錄本身的生成時間，Flink經過時間戳分配器訪問事件時間戳。vim
Ingestion Time：是數據進入Flink的時間。
Processing Time：是每個執行基於時間操做的算子的本地系統時間，與機器相關，默認的時間屬性就是Processing Time。

例如，一條日誌進入Flink的時間爲2021-01-22 10:00:00.123，到達Window的系統時間爲2021-01-22 10:00:01.234，日誌的內容以下：
2021-01-06 18:37:15.624 INFO Fail over to rm2

對於業務來講，要統計1min內的故障日誌個數，哪一個時間是最有意義的？—— eventTime，由於咱們要根據日誌的生成時間進行統計。

Window

Window，即窗口，咱們前面一直提到的邊界就是這裏的Window(窗口)。

官方解釋：流式計算是一種被設計用於處理無限數據集的數據處理引擎，而無限數據集是指一種不斷增加的本質上無限的數據集，而window是一種切割無限數據爲有限塊進行處理的手段。

因此Window是無限數據流處理的核心，Window將一個無限的stream拆分紅有限大小的」buckets」桶，咱們能夠在這些桶上作計算操做。

Window類型

本文剛開始提到，劃分窗口就兩種方式：

根據時間進行截取(time-driven-window)，好比每1分鐘統計一次或每10分鐘統計一次。
根據數據進行截取(data-driven-window)，好比每5個數據統計一次或每50個數據統計一次。

窗口類型

對於TimeWindow(根據時間劃分窗口)，能夠根據窗口實現原理的不一樣分紅三類：滾動窗口（Tumbling Window）、滑動窗口（Sliding Window）和會話窗口（Session Window）。

滾動窗口（Tumbling Windows）

將數據依據固定的窗口長度對數據進行切片。

特色：時間對齊，窗口長度固定，沒有重疊。

滾動窗口分配器將每一個元素分配到一個指定窗口大小的窗口中，滾動窗口有一個固定的大小，而且不會出現重疊。

例如：若是你指定了一個5分鐘大小的滾動窗口，窗口的建立以下圖所示：

滾動窗口

適用場景：適合作BI統計等（作每一個時間段的聚合計算）。

滑動窗口（Sliding Windows）

滑動窗口是固定窗口的更廣義的一種形式，滑動窗口由固定的窗口長度和滑動間隔組成。

特色：時間對齊，窗口長度固定，有重疊。

滑動窗口分配器將元素分配到固定長度的窗口中，與滾動窗口相似，窗口的大小由窗口大小參數來配置，另外一個窗口滑動參數控制滑動窗口開始的頻率。所以，滑動窗口若是滑動參數小於窗口大小的話，窗口是能夠重疊的，在這種狀況下元素會被分配到多個窗口中。

例如，你有10分鐘的窗口和5分鐘的滑動，那麼每一個窗口中5分鐘的窗口裏包含着上個10分鐘產生的數據，以下圖所示：

滑動窗口

適用場景：對最近一個時間段內的統計（求某接口最近5min的失敗率來決定是否要報警）。

會話窗口（Session Windows）

由一系列事件組合一個指定時間長度的timeout間隙組成，相似於web應用的session，也就是一段時間沒有接收到新數據就會生成新的窗口。

特色：時間無對齊。

session窗口分配器經過session活動來對元素進行分組，session窗口跟滾動窗口和滑動窗口相比，不會有重疊和固定的開始時間和結束時間的狀況，相反，當它在一個固定的時間週期內再也不收到元素，即非活動間隔產生，那個這個窗口就會關閉。一個session窗口經過一個session間隔來配置，這個session間隔定義了非活躍週期的長度，當這個非活躍週期產生，那麼當前的session將關閉而且後續的元素將被分配到新的session窗口中去。

會話窗口

Window API

TimeWindow

TimeWindow是將指定時間範圍內的全部數據組成一個window，一次對一個window裏面的全部數據進行計算（就是本文開頭說的對一個邊界內的數據進行計算）。

咱們以 紅綠燈路口經過的汽車數量 爲例子：

紅綠燈路口會有汽車經過，一共會有多少汽車經過，沒法計算。由於車流源源不斷，計算沒有邊界。

因此咱們統計每15秒鐘經過紅路燈的汽車數量，如第一個15秒爲2輛，第二個15秒爲3輛，第三個15秒爲1輛 …

tumbling-time-window (無重疊數據)

咱們使用 Linux 中的 nc 命令模擬數據的發送方

1.開啓發送端口，端口號爲9999
nc -lk 9999

2.發送內容（key 表明不一樣的路口，value 表明每次經過的車輛）
一次發送一行，發送的時間間隔表明汽車通過的時間間隔
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4

Flink 進行採集數據並計算：

object Window {
  def main(args: Array[String]): Unit = {
    //TODO time-window
    //1.建立運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2.定義數據流來源
    val text = env.socketTextStream("localhost", 9999)

    //3.轉換數據格式，text->CarWc
    case class CarWc(sensorId: Int, carCnt: Int)
    val ds1: DataStream[CarWc] = text.map {
      line => {
        val tokens = line.split(",")
        CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
      }
    }

    //4.執行統計操做，每一個sensorId一個tumbling窗口，窗口的大小爲5秒
    //也就是說，每5秒鐘統計一次，在這過去的5秒鐘內，各個路口經過紅綠燈汽車的數量。
    val ds2: DataStream[CarWc] = ds1
      .keyBy("sensorId")
      .timeWindow(Time.seconds(5))
      .sum("carCnt")

    //5.顯示統計結果
    ds2.print()

    //6.觸發流計算
    env.execute(this.getClass.getName)

  }
}

咱們發送的數據並無指定時間字段，因此Flink使用的是默認的 Processing Time，也就是Flink系統處理數據時的時間。

sliding-time-window (有重疊數據)

//1.建立運行環境
val env = StreamExecutionEnvironment.getExecutionEnvironment

//2.定義數據流來源
val text = env.socketTextStream("localhost", 9999)

//3.轉換數據格式，text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)

val ds1: DataStream[CarWc] = text.map {
  line => {
    val tokens = line.split(",")
    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
  }
}
//4.執行統計操做，每一個sensorId一個sliding窗口，窗口時間10秒,滑動時間5秒
//也就是說，每5秒鐘統計一次，在這過去的10秒鐘內，各個路口經過紅綠燈汽車的數量。
val ds2: DataStream[CarWc] = ds1
  .keyBy("sensorId")
  .timeWindow(Time.seconds(10), Time.seconds(5))
  .sum("carCnt")

//5.顯示統計結果
ds2.print()

//6.觸發流計算
env.execute(this.getClass.getName)

CountWindow

CountWindow根據窗口中相同key元素的數量來觸發執行，執行時只計算元素數量達到窗口大小的key對應的結果。

注意：CountWindow的window_size指的是相同Key的元素的個數，不是輸入的全部元素的總數。

tumbling-count-window (無重疊數據)

//1.建立運行環境
val env = StreamExecutionEnvironment.getExecutionEnvironment

//2.定義數據流來源
val text = env.socketTextStream("localhost", 9999)

//3.轉換數據格式，text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)

val ds1: DataStream[CarWc] = text.map {
  (f) => {
    val tokens = f.split(",")
    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
  }
}
//4.執行統計操做，每一個sensorId一個tumbling窗口，窗口的大小爲5
//按照key進行收集，對應的key出現的次數達到5次做爲一個結果
val ds2: DataStream[CarWc] = ds1
  .keyBy("sensorId")
  .countWindow(5)
  .sum("carCnt")

//5.顯示統計結果
ds2.print()

//6.觸發流計算
env.execute(this.getClass.getName)

sliding-count-window (有重疊數據)

一樣也是窗口長度和滑動窗口的操做：窗口長度是5，滑動長度是3

//1.建立運行環境
val env = StreamExecutionEnvironment.getExecutionEnvironment

//2.定義數據流來源
val text = env.socketTextStream("localhost", 9999)

//3.轉換數據格式，text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)

val ds1: DataStream[CarWc] = text.map {
  (f) => {
    val tokens = f.split(",")
    CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
  }
}
//4.執行統計操做，每一個sensorId一個sliding窗口，窗口大小3條數據,窗口滑動爲3條數據
//也就是說，每一個路口分別統計，收到關於它的3條消息時統計在最近5條消息中，各自路口經過的汽車數量
val ds2: DataStream[CarWc] = ds1
  .keyBy("sensorId")
  .countWindow(5, 3)
  .sum("carCnt")

//5.顯示統計結果
ds2.print()

//6.觸發流計算
env.execute(this.getClass.getName)

Window 總結

flink支持兩種劃分窗口的方式（time和count）

若是根據時間劃分窗口，那麼它就是一個time-window
若是根據數據劃分窗口，那麼它就是一個count-window

flink支持窗口的兩個重要屬性（size和interval）

若是size=interval,那麼就會造成tumbling-window(無重疊數據)
若是size>interval,那麼就會造成sliding-window(有重疊數據)
若是size<interval,那麼這種窗口將會丟失數據。好比每5秒鐘，統計過去3秒的經過路口汽車的數據，將會漏掉2秒鐘的數據。

經過組合能夠得出四種基本窗口

time-tumbling-window 無重疊數據的時間窗口，設置方式舉例：timeWindow(Time.seconds(5))
time-sliding-window 有重疊數據的時間窗口，設置方式舉例：timeWindow(Time.seconds(5), Time.seconds(3))
count-tumbling-window無重疊數據的數量窗口，設置方式舉例：countWindow(5)
count-sliding-window 有重疊數據的數量窗口，設置方式舉例：countWindow(5,3)

Window Reduce

WindowedStream → DataStream：給window賦一個reduce功能的函數，並返回一個聚合的結果。

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object StreamWindowReduce {
  def main(args: Array[String]): Unit = {
    // 獲取執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 建立SocketSource
    val stream = env.socketTextStream("node01", 9999)

    // 對stream進行處理並按key聚合
    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)

    // 引入時間窗口
    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))

    // 執行聚合操做
    val streamReduce = streamWindow.reduce(
      (item1, item2) => (item1._1, item1._2 + item2._2)
    )

    // 將聚合數據寫入文件
    streamReduce.print()

    // 執行程序
    env.execute("TumblingWindow")
  }
}

Window Apply

apply方法能夠進行一些自定義處理，經過匿名內部類的方法來實現。當有一些複雜計算時使用。

用法

實現一個 WindowFunction 類
指定該類的泛型爲 [輸入數據類型, 輸出數據類型, keyBy中使用分組字段的類型, 窗口類型]

示例：使用apply方法來實現單詞統計

步驟：

獲取流處理運行環境
構建socket流數據源，並指定IP地址和端口號
對接收到的數據轉換成單詞元組
使用 keyBy 進行分流（分組）
使用 timeWinodw 指定窗口的長度（每3秒計算一次）
實現一個WindowFunction匿名內部類

apply方法中實現聚合計算
使用Collector.collect收集數據

核心代碼以下：

    //1. 獲取流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2. 構建socket流數據源，並指定IP地址和端口號
    val textDataStream = env.socketTextStream("node01", 9999).flatMap(_.split(" "))

    //3. 對接收到的數據轉換成單詞元組
    val wordDataStream = textDataStream.map(_->1)

    //4. 使用 keyBy 進行分流（分組）
    val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream.keyBy(_._1)

    //5. 使用 timeWinodw 指定窗口的長度（每3秒計算一次）
    val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(3))

    //6. 實現一個WindowFunction匿名內部類
    val reduceDatStream: DataStream[(String, Int)] = windowDataStream.apply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
      //在apply方法中實現數據的聚合
      override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
        println("hello world")
        val tuple = input.reduce((t1, t2) => {
          (t1._1, t1._2 + t2._2)
        })
        //將要返回的數據收集起來，發送回去
        out.collect(tuple)
      }
    })
    reduceDatStream.print()
    env.execute()

Window Fold

WindowedStream → DataStream：給窗口賦一個fold功能的函數，並返回一個fold後的結果。

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object StreamWindowFold {
  def main(args: Array[String]): Unit = {
    // 獲取執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 建立SocketSource
    val stream = env.socketTextStream("node01", 9999,'\n',3)

    // 對stream進行處理並按key聚合
    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)

    // 引入滾動窗口
    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))

    // 執行fold操做
    val streamFold = streamWindow.fold(100){
      (begin, item) =>
        begin + item._2
    }

    // 將聚合數據寫入文件
    streamFold.print()

    // 執行程序
    env.execute("TumblingWindow")
  }
}

Aggregation on Window

WindowedStream → DataStream：對一個window內的全部元素作聚合操做。min和 minBy的區別是min返回的是最小值，而minBy返回的是包含最小值字段的元素(一樣的原理適用於 max 和 maxBy)。

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._

object StreamWindowAggregation {
  def main(args: Array[String]): Unit = {
    // 獲取執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 建立SocketSource
    val stream = env.socketTextStream("node01", 9999)

    // 對stream進行處理並按key聚合
    val streamKeyBy = stream.map(item => (item.split(" ")(0), item.split(" ")(1))).keyBy(0)

    // 引入滾動窗口
    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))

    // 執行聚合操做
    val streamMax = streamWindow.max(1)

    // 將聚合數據寫入文件
    streamMax.print()

    // 執行程序
    env.execute("TumblingWindow")
  }
}

EventTime與Window

EventTime的引入

與現實世界中的時間是不一致的，在flink中被劃分爲事件時間，提取時間，處理時間三種。
若是以EventTime爲基準來定義時間窗口那將造成EventTimeWindow,要求消息自己就應該攜帶EventTime
若是以IngesingtTime爲基準來定義時間窗口那將造成IngestingTimeWindow,以source的systemTime爲準。
若是以ProcessingTime基準來定義時間窗口那將造成ProcessingTimeWindow，以operator的systemTime爲準。

在Flink的流式處理中，絕大部分的業務都會使用eventTime，通常只在eventTime沒法使用時，纔會被迫使用ProcessingTime或者IngestionTime。

若是要使用EventTime，那麼須要引入EventTime的時間屬性，引入方式以下所示：

val env = StreamExecutionEnvironment.getExecutionEnvironment

// 從調用時刻開始給env建立的每個stream追加時間特徵
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

Watermark

引入

咱們知道，流處理從事件產生，到流經 source，再到 operator，中間是有一個過程和時間的，雖然大部分狀況下，流到 operator 的數據都是按照事件產生的時間順序來的，可是也不排除因爲網絡、背壓等緣由，致使亂序的產生，所謂亂序，就是指 Flink 接收到的事件的前後順序不是嚴格按照事件的 Event Time 順序排列的，因此 Flink 最初設計的時候，就考慮到了網絡延遲，網絡亂序等問題，因此提出了一個抽象概念：水印（WaterMark）；

如上圖所示，就出現一個問題，一旦出現亂序，若是隻根據 EventTime 決定 Window 的運行，咱們不能明確數據是否所有到位，但又不能無限期的等下去，此時必需要有個機制來保證一個特定的時間後，必須觸發 Window 去進行計算了，這個特別的機制，就是 Watermark。

Watermark 是用於處理亂序事件的，而正確的處理亂序事件，一般用 Watermark 機制結合 Window 來實現。

數據流中的 Watermark 用於表示 timestamp 小於 Watermark 的數據，都已經到達了，所以，Window 的執行也是由 Watermark 觸發的。

Watermark 能夠理解成一個延遲觸發機制，咱們能夠設置 Watermark 的延時時長 t，每次系統會校驗已經到達的數據中最大的 maxEventTime，而後認定 EventTime 小於 maxEventTime - t 的全部數據都已經到達，若是有窗口的中止時間等於 maxEventTime – t，那麼這個窗口被觸發執行。

有序流的Watermarker以下圖所示：（Watermark設置爲0）

有序數據的Watermark

亂序流的Watermarker以下圖所示：（Watermark設置爲2）

無序數據的Watermark

當 Flink 接收到每一條數據時，都會產生一條 Watermark，這條 Watermark 就等於當前全部到達數據中的 maxEventTime - 延遲時長，也就是說，Watermark 是由數據攜帶的，一旦數據攜帶的 Watermark 比當前未觸發的窗口的中止時間要晚，那麼就會觸發相應窗口的執行。因爲 Watermark 是由數據攜帶的，所以，若是運行過程當中沒法獲取新的數據，那麼沒有被觸發的窗口將永遠都不被觸發。

上圖中，咱們設置的容許最大延遲到達時間爲2s，因此時間戳爲7s的事件對應的Watermark是5s，時間戳爲12s的事件的Watermark是10s，若是咱們的窗口1是1s~5s，窗口2是6s~10s，那麼時間戳爲7s的事件到達時的Watermarker剛好觸發窗口1，時間戳爲12s的事件到達時的Watermark剛好觸發窗口2。

Flink對於遲到數據的處理

waterMark和Window機制解決了流式數據的亂序問題，對於由於延遲而順序有誤的數據，能夠根據eventTime進行業務處理，於延遲的數據Flink也有本身的解決辦法，主要的辦法是給定一個容許延遲的時間，在該時間範圍內仍能夠接受處理延遲數據。

設置容許延遲的時間是經過 allowedLateness(lateness: Time) 設置

保存延遲數據則是經過 sideOutputLateData(outputTag: OutputTag[T]) 保存

獲取延遲數據是經過 DataStream.getSideOutput(tag: OutputTag[X]) 獲取

具體的用法以下：

allowedLateness(lateness: Time)

def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {
  javaStream.allowedLateness(lateness)
  this
}

該方法傳入一個Time值，設置容許數據遲到的時間，這個時間和 WaterMark 中的時間概念不一樣。再來回顧一下：

WaterMark=數據的事件時間-容許亂序時間值

隨着新數據的到來，waterMark的值會更新爲最新數據事件時間-容許亂序時間值，可是若是這時候來了一條歷史數據，waterMark值則不會更新。總的來講，waterMark是爲了能接收到儘量多的亂序數據。

那這裏的Time值，主要是爲了等待遲到的數據，在必定時間範圍內，若是屬於該窗口的數據到來，仍會進行計算，後面會對計算方式仔細說明

注意：該方法只針對於基於event-time的窗口，若是是基於processing-time，而且指定了非零的time值則會拋出異常。

sideOutputLateData(outputTag: OutputTag[T])

def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {
  javaStream.sideOutputLateData(outputTag)
  this
}

該方法是將遲來的數據保存至給定的outputTag參數，而OutputTag則是用來標記延遲數據的一個對象。

DataStream.getSideOutput(tag: OutputTag[X])

經過window等操做返回的DataStream調用該方法，傳入標記延遲數據的對象來獲取延遲的數據。

對延遲數據的理解

延遲數據是指：

在當前窗口【假設窗口範圍爲10-15】已經計算以後，又來了一個屬於該窗口的數據【假設事件時間爲13】，這時候仍會觸發 Window 操做，這種數據就稱爲延遲數據。

那麼問題來了，延遲時間怎麼計算呢？

假設窗口範圍爲10-15，延遲時間爲2s，則只要 WaterMark<15+2，而且屬於該窗口，就能觸發 Window 操做。而若是來了一條數據使得 WaterMark>=15+2，10-15這個窗口就不能再觸發 Window 操做，即便新來的數據的 Event Time 屬於這個窗口時間內。

Flink 關聯 Hive 分區表

Flink 1.12 支持了 Hive 最新的分區做爲時態表的功能，能夠經過 SQL 的方式直接關聯 Hive 分區表的最新分區，而且會自動監聽最新的 Hive 分區，當監控到新的分區後，會自動地作維表數據的全量替換。經過這種方式，用戶無需編寫 DataStream 程序便可完成 Kafka 流實時關聯最新的 Hive 分區實現數據打寬。

具體用法：

在 Sql Client 中註冊 HiveCatalog：

vim conf/sql-client-defaults.yaml 
catalogs: 
  - name: hive_catalog 
    type: hive 
    hive-conf-dir: /disk0/soft/hive-conf/ #該目錄須要包hive-site.xml文件

建立 Kafka 表

CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 (  
    master Row<reportDate String, groupID int, shopID int, shopName String, action int, orderStatus int, orderKey String, actionTime bigint, areaName String, paidAmount double, foodAmount double, startTime String, person double, orderSubType int, checkoutTime String>,  
proctime as PROCTIME()  -- PROCTIME用來和Hive時態表關聯  
) WITH (  
 'connector' = 'kafka',  
 'topic' = 'topic_name',  
 'format' = 'json',  
 'properties.bootstrap.servers' = 'host:9092',  
 'properties.group.id' = 'flinkTestGroup',  
 'scan.startup.mode' = 'timestamp',  
 'scan.startup.timestamp-millis' = '1607844694000'  
);

Flink 事實表與 Hive 最新分區數據關聯

dim_extend_shop_info 是 Hive 中已存在的表，因此咱們用 table hint 動態地開啓維表參數。

CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as  
SELECT * FROM  
 (select t1.*, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id,   
     ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn  
    from hive_catalog.flink_db.kfk_fact_bill_master_12 t1  
       JOIN hive_catalog.flink_db.dim_extend_shop_info   
  /*+ OPTIONS('streaming-source.enable'='true',  
     'streaming-source.partition.include' = 'latest',  
     'streaming-source.monitor-interval' = '1 h',
     'streaming-source.partition-order' = 'partition-name') */
    FOR SYSTEM_TIME AS OF t1.proctime AS t2 --時態表  
    ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id  
    where groupID in (202042)) t  where t.rn = 1

參數解釋：

streaming-source.enable 開啓流式讀取 Hive 數據。
streaming-source.partition.include 有如下兩個值：

latest 屬性: 只讀取最新分區數據。
all: 讀取全量分區數據，默認值爲 all，表示讀全部分區，latest 只能用在 temporal join 中，用於讀取最新分區做爲維表，不能直接讀取最新分區數據。

streaming-source.monitor-interval 監聽新分區生成的時間、不宜太短、最短是1 個小時，由於目前的實現是每一個 task 都會查詢 metastore，高頻的查可能會對metastore 產生過大的壓力。須要注意的是，1.12.1 放開了這個限制，但仍建議按照實際業務不要配個過短的 interval。

streaming-source.partition-order 分區策略，主要有如下 3 種，其中最爲推薦的是 partition-name：