Flink watermark

Flink中watermark主要解決保序問題. 而保序問題的根本緣由是多個任務同時從流中並行處理數據，順序沒法保證.

上游: 生成watermark
通常在WINDOW 操做以前生成WATERMARK, WATERMARK 有兩種:
AssignWithPeriodicWatermarks:
每隔N秒自動向流裏注入一個WATERMARK 時間間隔由ExecutionConfig.setAutoWatermarkInterval 決定. 每次調用getCurrentWatermark 方法, 若是獲得的WATERMARK 不爲空而且比以前的大就注入流中 (emitWatermark)
參考 TimestampsAndPeriodicWatermarksOperator.processElement

AssignWithPunctuatedWatermarks:
基於事件向流裏注入一個WATERMARK，每個元素都有機會判斷是否生成一個WATERMARK. 若是獲得的WATERMARK 不爲空而且比以前的大就注入流中 (emitWatermark)
參考 TimestampsAndPunctuatedWatermarksOperator.processElement

每次生成WATERMARK將覆蓋流中已有的WATERMARK

下游: 處理watermark
StatusWatermarkValve 負責將不一樣Channel 的Watermark 對齊，再傳給pipeline 下游,對齊的概念是當前Channel的Watermark時間大於全部Channel最小的Watermark時間

WindowOperator 的處理:
WindowOperator.processElement

1. WindowAssigner.assignWindows 爲當前的消息分配滑動窗口
   經常使用的有： TumblingEventTimeWindows: 按照消息的 EventTime 分配窗口 (每次生成單個窗口)
   TumblingProcessingTimeWindows 按照當前的時間分配窗口 (每次生成單個窗口)
   須要配合StreamExecutionEnvironment.setStreamTimeCharacteristic 使用 (默認是TimeCharacteristic.ProcessingTime), 這個必須匹配
   不然沒法正常觸發滑動窗口

實際觀察結果：

• 若是使用ProcessingTimeWindows 即便Event 自己的時間落後於窗口時間不少也會被觸發
• 不管是否使用WATERMARK，窗口中的數據會有亂序，即後到窗口中的數據早於先到窗口中的數據
• 若是使用EventTimeWindow, 數據和窗口時間對齊不會亂序，同一窗口中的數據不能嚴格保證順序，須要SORT.
• 最後一批數據有缺失，缺失的數據取決於WATERMARK的MAXOUTOFORDERNESS
• 默認的WATERMARK算法是根據元素的最大時間決定的，當沒有新的元素進入流中的時候，水位再也不上漲，再減去MAXOUTOFORDERNESS, 則最後一批數據沒法落在水位之下，致使WINDOW沒法觸發

2. 將當前的滑動窗口和對象加入WindowState, 根據不一樣的應用場景會使用不一樣的WindowState. WindowState 的類型由WindowedStream的具體操做決定, 生成對應的StateDescriptor, 不一樣的WindowState 的 add/get 行爲會不一樣. 好比HeapListWindowState 會把當前的對象追加到currentNamespace (即Timewindow) 對應的List 下. 好比HeapAggregateState 會對當前的對象應用Aggregate function 並更新結果

Window 觸發的條件
在 WindowOperator 中有兩個點會檢查窗口是否觸發，二者的檢查條件有所不一樣

1. processElement 這是在新的流數據進入時觸發
   檢查條件： watermark時間 >= 窗口最大時間 參見 EventTimeTrigger.onElement
   若是窗口不能被觸發則調用InteralTimeService.registerEventTimeTimer 註冊一個定時器,以KEY+窗口最大時間爲條件觸發, 到必定時間後定時器會被自動銷燬. 時間爲窗口最大時間+WindowOperator.allowedLateness WindowOperator.allowedLateness 能夠經過 Stream.window(...).allowedLateness(...) 設置. 通常應該略大於WatermarkGenerator 的 maxOutOfOrderness

2. onEventTime 或者 onProcessingTime 取決於Watermark的類型, 這是在Watermark更新的時候觸發 （InteralTimeService.advanceWatermark). 這時會把當前Watermark 的時間和以前註冊的定時器的時間作比較, 若是定時器還存在而且Watermark的時間大於定時器時間則能夠觸發窗口. 參見 EventTimeTrigger.onEventTime
   
   參考 http://blog.csdn.net/lmalds/article/details/52704170

WATERMARK和普通數據分開處理
若是一個元素來的過晚 element.getTimestamp + allowedLateness < currentWatermark
會有一個特殊的OutputTag 和正常的流數據區分開
參考 https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/side_output.html

若是窗口來的過晚, window.maxTimestamp + allowedLateness < currentWatermark, 則窗口會被直接丟棄

Watermark 的問題：
默認的Watermark機制是數據驅動的，新的數據進入纔會觸發水位上升, 而因爲maxOutOfOrderness 的存在, watermark < 最大流數據時間 < 當前窗口結束時間
根據以前的分析，最新的時間窗口老是不會被觸發，除非更新的數據進入再次提升水位到當前窗口結束時間之後, 若是數據進入的頻率低或者沒有新的數據進入流，那最新的時間窗口被處理的延時會很是高甚至永遠不會被觸發，這在實時性要求高的流式系統是很致命的. 好比一個銀行系統，要作客戶帳號層面的保序，每一個帳號的交易可能一天只有幾筆甚至一筆，若是咱們在Window 處理的時候KEY BY 帳號就會引發上述問題. 咱們能夠考慮KEY BY的條件改成 HASH(帳號) 再取模，而後在窗口處理中再次根據帳號分組，這樣雖然處理複雜一些，可是保證了窗口中數據的頻次

另一種方案是優化WATERMARK生成的機制，若是一段時間後WATERMARK仍然沒有變化，那就將WATERMARK自動上漲一次到當前窗口的結束時間，這樣保證窗口處理的延時有個上限

public abstract class AbstractWatermarkGenerator<T> implements AssignerWithPeriodicWatermarks<T> {
    private static final long serialVersionUID = -2006930231735705083L;
    private static final Logger logger = LoggerFactory.getLogger(AbstractWatermarkGenerator.class);
    private final long maxOutOfOrderness; // 10 seconds
    private long windowSize;
    private long currentMaxTimestamp;
    private long lastTimestamp = 0;
    private long lastWatermarkChangeTime = 0;
    private long windowPurgeTime = 0;

    public AbstractWatermarkGenerator(long maxOutOfOrderness, long windowSize) {
        this.maxOutOfOrderness = maxOutOfOrderness;
        this.windowSize = windowSize;
    }

    public AbstractWatermarkGenerator() {
        this(10000, 10000);
    }

    protected abstract long extractCurTimestamp(T element) throws Exception;

    public long extractTimestamp(T element,
            long previousElementTimestamp) {
        try {
            long curTimestamp = extractCurTimestamp(element);
            lastWatermarkChangeTime = new Date().getTime();
            currentMaxTimestamp = Math.max(curTimestamp, currentMaxTimestamp);
            windowPurgeTime = Math.max(windowPurgeTime, getWindowExpireTime(currentMaxTimestamp));
            if (logger.isDebugEnabled()) {
                logger.debug("Extracting timestamp: {}", currentMaxTimestamp);
            }
            return curTimestamp;
        } catch (Exception e) {
            logger.error("Error extracting timestamp", e);      
        }

        return 0;
    }

    protected long getWindowExpireTime(long currentMaxTimestamp) {
        long windowStart = TimeWindow.getWindowStartWithOffset(currentMaxTimestamp, 0, windowSize);
        long windowEnd = windowStart + windowSize;
        return windowEnd + maxOutOfOrderness;
    }

    public Watermark getCurrentWatermark() {
        long curTime = new Date().getTime();
        if (currentMaxTimestamp > lastTimestamp) {
            if (logger.isDebugEnabled()) {
                logger.debug("Current max timestamp has been increased since last");
            }
            lastTimestamp = currentMaxTimestamp;
            lastWatermarkChangeTime = curTime;
        }
        else {
            long diff = windowPurgeTime - currentMaxTimestamp;
            if (diff > 0 && curTime - lastWatermarkChangeTime > diff) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Increase current MaxTimestamp once");
                }
                currentMaxTimestamp = windowPurgeTime;
                lastTimestamp = currentMaxTimestamp;
                lastWatermarkChangeTime = curTime;
            }
        }

        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

實際測試中發現 WATERMARK是否觸發和算子的併發度和WATERMARK生成的位置有關
測試結果以下：

• Env default parallism 10: Source parallism 20, window parallism 6, watermark 生成定義在keyby 以前
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲10， WINDOW SUBTASK 併發度爲6, 窗口能夠正常觸發
• Env default parallism 20, Source parallism 20, window parallism 6, watermark 生成定義在keyby 以前
  Source 到 WINDOW 算子以前 合成一個SUBTASK，併發度爲20， WINDOW SUBTASK 併發度爲6, 窗口能夠正常觸發
• Env default parallism 60, Source parallism 20, window parallism 10, watermark 生成定義在keyby 以前
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲60，WINDOW SUBTASK 併發度爲10, 窗口不能正常觸發 (我的理解緣由是算子併發度擴大，致使一些CHANNEL處理線程沒有數據，根據上文的解釋，WATERMARK對齊會取全部CHANNEL最小的WATERMARK，致使水位沒法上漲
  能夠從FLINK CONSOLE的WATERMARKS看出)
• Env default parallism 60, Source parallism 20, window parallism 10, watermark 生成定義在Source以後
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲60，WINDOW SUBTASK 併發度爲10, 窗口能夠正常觸發
• Env default parallism 10, Source parallism 20, window parallism 20, watermark 生成定義在keyby 以前
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲10， WINDOW SUBTASK 併發度爲20, 窗口能夠正常觸發
• Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在keyby 以前
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲30， WINDOW SUBTASK 併發度爲20, 窗口不能正常觸發
• Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在keyby 以前
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲30， WINDOW SUBTASK 併發度爲20, 窗口不能正常觸發
• Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在Source 以後
  Source 爲單獨的SUBTASK 併發度爲20, 以後到WINDOW算子以前合成一個SUBTASK，併發度爲30， WINDOW SUBTASK 併發度爲20, 窗口能夠正常觸發

因此注意WINDOW算子以前最好避免讓下游算子的併發度超過上游算子，不然就把WATERMARK的生成儘可能放到DAG的前端，這樣WATERMARK能夠被傳遞到下游算子