Flink基礎：時間和水印

時間  2020-11-15
標籤 java sql 緩存 ide 函數 源碼分析 3d 日誌 code 對象 欄目 Java 简体版
原文   原文鏈接

​

往期推薦：java

Flink基礎：入門介紹sql

Flink基礎：DataStream API緩存

Flink基礎：實時處理管道與ETLide

Flink深刻淺出：資源管理函數

Flink深刻淺出：部署模式源碼分析

Flink深刻淺出：內存模型3d

Flink深刻淺出：JDBC Source從理論到實戰日誌

Flink深刻淺出：Sql Gateway源碼分析code

Flink深刻淺出：JDBC Connector源碼分析對象

 

本篇終於到了Flink的核心內容：時間與水印。最初接觸這個概念是在Spark Structured Streaming中，一直沒法理解水印的做用。直到使用了一段時間Flink以後，對實時流處理有了必定的理解，纔想清楚其中的原因。接下來就來介紹下Flink中的時間和水印，以及基於時間特性支持的窗口處理。

1 時間和水印

1.1 介紹

Flink支持不一樣的時間類型：

  • 事件時間：事件發生的時間，是設備生產或存儲事件的時間，通常都直接存儲在事件上，好比Mysql Binglog中的修改時間；或者用戶訪問日誌的訪問時間等。
  • 攝入時間：事件進入Flink的時間，這個時間不經常使用。
  • 處理時間：某個特殊的算子處理事件的時間，當不在乎事件的順序時，爲了保證高吞吐低延遲，會採用這種時間。

好比想要計算給定某天的第一個小時的股票價格趨勢，就須要使用事件時間。若是選擇處理時間進行計算，那麼將會按照當前Flink應用處理的時間進行統計，就可能會形成數據一致性問題，歷史數據的分析也很難復現。還有個典型的場景是流式處理每每是7*24小時不間斷的運行，加入使用處理時間，當中間停機進行代碼更新或者BUG處理時，再次啓動，中間未處理的數據會堆積當重啓時間一次性處理，這樣對統計結果就形成大大的干擾。

1.2 使用EventTime

Flink默認使用的是處理時間，能夠經過下面的方法修改爲事件時間：

final StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

若是須要使用事件時間，還須要提供時間抽取器和水印生成器，這樣Flink才能夠追蹤到事件時間的處理進度。

1.3 水印

經過下面的例子，能夠了解爲何須要水印，水印是怎麼工做的。在這個例子中，每一個事件都帶有一個時間標識，下面的數字就是事件上的時間，很明顯它們是亂序到達的。第一個到達的是4，而後是2：

23 19 22 24 21 14 17 13 12 15 9 11 7 2 4(第一個事件) 

加入如今但願對流進行排序，那麼每一個事件到達的時候，就須要產生一個流，按照時間戳排好序輸出每一個到達的事件。

  • 上帝視角：第一個到達的事件是4，可是不能馬上就把它當作第一個元素放入排序流中，由於如今事件是亂序的，沒法肯定前面的事件是否已經到達。固然如今你已經看到完整的事件順序，固然會知道只要再等待一個事件，4以前的事件就都處理完了（這就是上帝視角），但在現實中咱們是一條條接收的數據，沒法知道4後面出現的是2。
  • 緩存和延遲：若是使用緩存，那麼頗有可能會永遠中止等待。第一個事件是4，第二個事件是2，咱們是否是隻須要等待一個事件就能保證事件的完整？多是，也可能不是，好比如今事件就永遠等待不到1。
  • 排序策略：對於任何給定的時間事件中止等待以前的數據，直接進行排序。這就是水印的做用：用來定義什麼時候中止等待更早的數據。Flink中的事件時間處理依賴於水印生成器，每當元素進入到Flink，會根據其事件時間，生成一個新的時間戳，即水印。對於t時間的水印，意味着Flink不會再接收t以前的數據，那麼t以前的數據就能夠進行排序產出順序流了。在上面的例子中，當水印的時間戳到達2時，就會把2事件輸出。
  • 水印策略：每當事件延遲到達時，這些延遲都不是固定的，一種簡單的方式是按照最大的延遲事件來判斷。對於大部分的應用，這種固定水印均可以工做的比較好。
1.4 延遲和完整性

在批處理中，用戶能夠一次性看到所有的數據，所以能夠很容易的知道事件的順序。在流處理中總須要等待一段時間，肯定事件完整後才能產生結果。能夠很激進的配置一個較短的水印延遲時間，這樣雖然輸入結果不完整（有的時間延遲還未到達就已經開始計算），可是速度會很快。或者設置較長的延遲，數據會相對完整，可是會有必定的延遲。也能夠採用混合的策略，剛開始延遲小一點，當處理了部分數據後，延遲增長。

1.5 延時

延時經過水印來定義，Watermark(t)表明了t時間的事件是完整的，即小於t的事件均可以開始處理了。

1.6 使用水印

爲了支撐事件時間機制的處理，Flink須要知道每一個事件的時間，而後爲其產生一個水印。

DataStream<Event> stream = ...

WatermarkStrategy<Event> strategy = WatermarkStrategy
  .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(20))
  // 選擇時間字段    
  .withTimestampAssigner((event, timestamp) -> event.timestamp);

DataStream<Event> withTimestampsAndWatermarks =
  // 定義水印生成的策略
  stream.assignTimestampsAndWatermarks(strategy);

2 窗口

Flink擁有豐富的窗口語義，接下來將會了解到：

  • 如何在無限數據流上使用窗口聚合數據
  • Flink都支持什麼類型的窗口
  • 如何實現一個窗口聚合
2.1 介紹

當進行流處理時很天然的想針對一部分數據聚合分析，好比想要統計每分鐘有多少瀏覽、每週每一個用戶有多少次會話、每分鐘每一個傳感器的最大溫度等。Flink的窗口分析依賴於兩個抽象概念：窗口分配器Assigner（用來指定事件屬於哪一個窗口，在必要的時候新建窗口），窗口函數Function（應用於窗口內的數據）。Flink的窗口也有觸發器Trigger的概念，它決定了什麼時候調用窗口函數進行處理；Evictor用於剔除窗口中不須要計算的數據。能夠像下面這樣建立窗口：

stream.
  .keyBy(<key selector>)
  .window(<window assigner>)
  .reduce|aggregate|process(<window function>)

也能夠在非key數據流上使用窗口，可是必定要當心，由於處理過程將不會並行執行：

stream.
  .windowAll(<window assigner>)
  .reduce|aggregate|process(<window function>)
2.2 窗口分配器

Flink有幾種內置的窗口分配器:

按照窗口聚合的種類能夠大體分爲：

  • 滾動窗口：好比統計每分鐘的瀏覽量，TumblingEventTimeWindows.of(Time.minutes(1))
  • 滑動窗口：好比每10秒鐘統計一次一分鐘內的瀏覽量，SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10))
  • 會話窗口：統計會話內的瀏覽量，會話的定義是同一個用戶兩次訪問不超過30分鐘，EventTimeSessionWindows.withGap(Time.minutes(30))

窗口的時間能夠經過下面的幾種時間單位來定義：

  • 毫秒，Time.milliseconds(n)
  • 秒，Time.seconds(n)
  • 分鐘，Time.minutes(n)
  • 小時，Time.hours(n)
  • 天，Time.days(n)

基於時間的窗口分配器支持事件時間和處理時間，這兩種類型的時間處理的吞吐量會有差異。使用處理時間優勢是延遲很低，可是也存在幾個缺點：沒法正確的處理歷史數據；沒法處理亂序數據；結果非冪等。當使用基於數量的窗口，若是數量不夠，可能永遠不會觸發窗口操做。沒有選項支持超時處理或部分窗口的處理，固然你能夠經過自定義窗口的方式來實現。全局窗口分配器會在一個窗口內，統一分配每一個事件。若是須要自定義窗口，通常會基於它來作。不過推薦直接使用ProcessFunction。

2.3 窗口函數

有三種選擇來處理窗口中的內容：

  • 當作批處理，使用ProcessWindowFunction，基於Iterable處理窗口內容
  • 增量的使用ReduceFunction和AggregateFunction依次處理窗口的每一個數據
  • 上面二者結合，使用ReduceFunction和AggregateFunction進行預聚合，而後使用ProcessFunction進行批量處理。

下面給出了方法1和方法3的例子，需求爲在每分鐘內尋找到每一個傳感器的值，產生<key,>的結果流。

2.3.1 ProcessWindowFunction的例子
DataStream<SensorReading> input = ...

input
    .keyBy(x -> x.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .process(new MyWastefulMax());

public static class MyWastefulMax extends ProcessWindowFunction<
        SensorReading,                  // input type
        Tuple3<String, Long, Integer>,  // output type
        String,                         // key type
        TimeWindow> {                   // window type

    @Override
    public void process(
            String key,
            Context context, 
            Iterable<SensorReading> events,
            Collector<Tuple3<String, Long, Integer>> out) {

        int max = 0;
        for (SensorReading event : events) {
            max = Math.max(event.value, max);
        }
        out.collect(Tuple3.of(key, context.window().getEnd(), max));
    }
}

有一些內容須要瞭解：

  • 全部窗口分配的時間都在Flink中按照key緩存起來，直到窗口觸發，所以代價很昂貴。

  • ProcessWindowFunction中傳入了Context對象，內部包含了對應的窗口信息，接口相似：

public abstract class Context implements java.io.Serializable {
    public abstract W window();

    public abstract long currentProcessingTime();
    public abstract long currentWatermark();

    public abstract KeyedStateStore windowState();
    public abstract KeyedStateStore globalState();
}

其中windowState和globalState會爲每一個key、每一個窗口或者全局存儲信息，當須要記錄窗口的某些信息的時候會頗有用。

2.3.2 Incremental Aggregation例子
DataStream<SensorReading> input = ...

input
    .keyBy(x -> x.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(new MyReducingMax(), new MyWindowFunction());

private static class MyReducingMax implements ReduceFunction<SensorReading> {
    public SensorReading reduce(SensorReading r1, SensorReading r2) {
        return r1.value() > r2.value() ? r1 : r2;
    }
}

private static class MyWindowFunction extends ProcessWindowFunction<
    SensorReading, Tuple3<String, Long, SensorReading>, String, TimeWindow> {

    @Override
    public void process(
            String key,
            Context context,
            Iterable<SensorReading> maxReading,
            Collector<Tuple3<String, Long, SensorReading>> out) {

        SensorReading max = maxReading.iterator().next();
        out.collect(Tuple3.of(key, context.window().getEnd(), max));
    }
}

注意iterable只會執行一次，即只有MyReducingMax輸出的值纔會傳入這裏。

2.4 延遲事件

默認當使用基於事件時間窗口時，延遲事件會直接丟棄。有兩種方法能夠處理這個問題：你能夠把須要丟棄的事件從新蒐集起來輸出到另外一個流中，也叫側輸出；或者配置水印的延遲時間。

OutputTag<Event> lateTag = new OutputTag<Event>("late"){};

SingleOutputStreamOperator<Event> result = stream.
  .keyBy(...)
  .window(...)
  .sideOutputLateData(lateTag)
  .process(...);

DataStream<Event> lateStream = result.getSideOutput(lateTag);

經過指定容許延遲的間隔時間，當在容許的延遲範圍內，仍然能夠分配到對應的窗口(窗口對應的狀態信息將會保留一段時間)。可是會致使對應窗口從新計算（也叫作延遲響應late firing）默認容許的延遲是0，也就是說一旦事件在水印以後就會被丟棄掉。

stream.
    .keyBy(...)
    .window(...)
    .allowedLateness(Time.seconds(10))
    .process(...);

當配置延遲後，只有那些在容許的延遲以外的數據會被丟棄或者使用側輸出蒐集起來。

3 注意

Flink的窗口處理可能跟你想的不太同樣，基於在flink用戶郵件中常問的問題，整理以下

3.1 滑動窗口形成數據拷貝

滑動窗口會形成大量的窗口對象，而且會拷貝每一個對象到對應的窗口中。好比，你的滑動窗口爲每15分鐘統計24小時的窗口長度，那麼每一個時間將會複製到4*24=96個窗口中。

3.2 時間窗口會對齊到系統時間

若是使用1個小時的窗口，那麼當應用在12:05啓動時，並非說第一個窗口的時間範圍是到1:05，事實上第一個窗口的時間是12:05到01:00，只有55分鐘而已。注意，滾動窗口和滑動窗口都支持偏移值的參數配置。

3.3 窗口後面能夠接窗口

好比：

stream
    .keyBy(t -> t.key)
    .timeWindow(<time specification>)
    .reduce(<reduce function>)
    .timeWindowAll(<same time specification>)
    .reduce(<same reduce function>)

這樣的代碼可以工做主要是由於第一個窗口輸出的內容系統會自動添加一個窗口結束的時間，後面的處理能夠基於這個時間再次進行窗口操做，可是須要窗口的配置統一或者整數倍。

3.4 空窗口沒有輸出

只有對應的事件到達時，纔會建立對應的窗口。所以若是沒有對應的事件，窗口就不會建立，所以也不會有任何輸出。

3.5 延遲數據形成延遲合併

對於會話窗口，實際上會爲每一個事件在一開始分配一個新的窗口，當新的事件到達時，會根據時間間隔合併窗口。所以若是事件延遲到達，頗有可能會形成窗口的延遲合併。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程