Flinkwatermark

Flink中watermark主要解決保序問(wèn)題. 而保序問(wèn)題的根本原因是多個(gè)任務(wù)同時(shí)從流中并行處理數(shù)據(jù)，順序無(wú)法保證.

網(wǎng)站建設(shè)哪家好，找創(chuàng)新互聯(lián)建站！專(zhuān)注于網(wǎng)頁(yè)設(shè)計(jì)、網(wǎng)站建設(shè)、微信開(kāi)發(fā)、微信小程序開(kāi)發(fā)、集團(tuán)企業(yè)網(wǎng)站建設(shè)等服務(wù)項(xiàng)目。為回饋新老客戶(hù)創(chuàng)新互聯(lián)還提供了柳城免費(fèi)建站歡迎大家使用！

上游: 生成watermark
一般在WINDOW 操作之前生成WATERMARK, WATERMARK 有兩種:
AssignWithPeriodicWatermarks:
每隔N秒自動(dòng)向流里注入一個(gè)WATERMARK 時(shí)間間隔由ExecutionConfig.setAutoWatermarkInterval 決定. 每次調(diào)用getCurrentWatermark 方法, 如果得到的WATERMARK 不為空并且比之前的大就注入流中 (emitWatermark)
參考 TimestampsAndPeriodicWatermarksOperator.processElement

AssignWithPunctuatedWatermarks:
基于事件向流里注入一個(gè)WATERMARK，每一個(gè)元素都有機(jī)會(huì)判斷是否生成一個(gè)WATERMARK. 如果得到的WATERMARK 不為空并且比之前的大就注入流中 (emitWatermark)
參考 TimestampsAndPunctuatedWatermarksOperator.processElement

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

下游: 處理watermark
StatusWatermarkValve 負(fù)責(zé)將不同Channel 的Watermark 對(duì)齊，再傳給pipeline 下游,對(duì)齊的概念是當(dāng)前Channel的Watermark時(shí)間大于所有Channel最小的Watermark時(shí)間
Flink watermark

WindowOperator 的處理:
WindowOperator.processElement

WindowAssigner.assignWindows 為當(dāng)前的消息分配滑動(dòng)窗口
常用的有： TumblingEventTimeWindows: 按照消息的 EventTime 分配窗口 (每次生成單個(gè)窗口)
TumblingProcessingTimeWindows 按照當(dāng)前的時(shí)間分配窗口 (每次生成單個(gè)窗口)
需要配合StreamExecutionEnvironment.setStreamTimeCharacteristic 使用 (默認(rèn)是TimeCharacteristic.ProcessingTime), 這個(gè)必須匹配
否則無(wú)法正常觸發(fā)滑動(dòng)窗口

實(shí)際觀察結(jié)果：

如果使用ProcessingTimeWindows 即使Event 本身的時(shí)間落后于窗口時(shí)間很多也會(huì)被觸發(fā)
無(wú)論是否使用WATERMARK，窗口中的數(shù)據(jù)會(huì)有亂序，即后到窗口中的數(shù)據(jù)早于先到窗口中的數(shù)據(jù)
如果使用EventTimeWindow, 數(shù)據(jù)和窗口時(shí)間對(duì)齊不會(huì)亂序，同一窗口中的數(shù)據(jù)不能?chē)?yán)格保證順序，需要SORT.
最后一批數(shù)據(jù)有缺失，缺失的數(shù)據(jù)取決于WATERMARK的MAXOUTOFORDERNESS
默認(rèn)的WATERMARK算法是根據(jù)元素的最大時(shí)間決定的，當(dāng)沒(méi)有新的元素進(jìn)入流中的時(shí)候，水位不再上漲，再減去MAXOUTOFORDERNESS, 則最后一批數(shù)據(jù)無(wú)法落在水位之下，導(dǎo)致WINDOW無(wú)法觸發(fā)

將當(dāng)前的滑動(dòng)窗口和對(duì)象加入WindowState, 根據(jù)不同的應(yīng)用場(chǎng)景會(huì)使用不同的WindowState. WindowState 的類(lèi)型由WindowedStream的具體操作決定, 生成對(duì)應(yīng)的StateDescriptor, 不同的WindowState 的 add/get 行為會(huì)不同. 比如HeapListWindowState 會(huì)把當(dāng)前的對(duì)象追加到currentNamespace (即Timewindow) 對(duì)應(yīng)的List 下. 比如HeapAggregateState 會(huì)對(duì)當(dāng)前的對(duì)象應(yīng)用Aggregate function 并更新結(jié)果

Window 觸發(fā)的條件
在 WindowOperator 中有兩個(gè)點(diǎn)會(huì)檢查窗口是否觸發(fā)，兩者的檢查條件有所不同

processElement 這是在新的流數(shù)據(jù)進(jìn)入時(shí)觸發(fā)
檢查條件： watermark時(shí)間 >= 窗口最大時(shí)間參見(jiàn) EventTimeTrigger.onElement
如果窗口不能被觸發(fā)則調(diào)用InteralTimeService.registerEventTimeTimer 注冊(cè)一個(gè)定時(shí)器,以KEY+窗口最大時(shí)間為條件觸發(fā), 到一定時(shí)間后定時(shí)器會(huì)被自動(dòng)銷(xiāo)毀. 時(shí)間為窗口最大時(shí)間+WindowOperator.allowedLateness WindowOperator.allowedLateness 可以通過(guò) Stream.window(...).allowedLateness(...) 設(shè)置. 一般應(yīng)該略大于WatermarkGenerator 的 maxOutOfOrderness
onEventTime 或者 onProcessingTime 取決于Watermark的類(lèi)型, 這是在Watermark更新的時(shí)候觸發(fā) （InteralTimeService.advanceWatermark). 這時(shí)會(huì)把當(dāng)前Watermark 的時(shí)間和之前注冊(cè)的定時(shí)器的時(shí)間做比較, 如果定時(shí)器還存在并且Watermark的時(shí)間大于定時(shí)器時(shí)間則可以觸發(fā)窗口. 參見(jiàn) EventTimeTrigger.onEventTime

參考 http://blog.csdn.net/lmalds/article/details/52704170

WATERMARK和普通數(shù)據(jù)分開(kāi)處理
如果一個(gè)元素來(lái)的過(guò)晚 element.getTimestamp + allowedLateness < currentWatermark
會(huì)有一個(gè)特殊的OutputTag 和正常的流數(shù)據(jù)區(qū)分開(kāi)
參考 https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/side_output.html

如果窗口來(lái)的過(guò)晚, window.maxTimestamp + allowedLateness < currentWatermark, 則窗口會(huì)被直接丟棄

Watermark 的問(wèn)題：
默認(rèn)的Watermark機(jī)制是數(shù)據(jù)驅(qū)動(dòng)的，新的數(shù)據(jù)進(jìn)入才會(huì)觸發(fā)水位上升, 而由于maxOutOfOrderness 的存在, watermark < 最大流數(shù)據(jù)時(shí)間 < 當(dāng)前窗口結(jié)束時(shí)間
根據(jù)之前的分析，最新的時(shí)間窗口總是不會(huì)被觸發(fā)，除非更新的數(shù)據(jù)進(jìn)入再次提高水位到當(dāng)前窗口結(jié)束時(shí)間以后, 如果數(shù)據(jù)進(jìn)入的頻率低或者沒(méi)有新的數(shù)據(jù)進(jìn)入流，那最新的時(shí)間窗口被處理的延時(shí)會(huì)非常高甚至永遠(yuǎn)不會(huì)被觸發(fā)，這在實(shí)時(shí)性要求高的流式系統(tǒng)是很致命的. 比如一個(gè)銀行系統(tǒng)，要做客戶(hù)賬號(hào)層面的保序，每個(gè)賬號(hào)的交易可能一天只有幾筆甚至一筆，如果我們?cè)赪indow 處理的時(shí)候KEY BY 賬號(hào)就會(huì)引起上述問(wèn)題. 我們可以考慮KEY BY的條件改為 HASH(賬號(hào)) 再取模，然后在窗口處理中再次根據(jù)賬號(hào)分組，這樣雖然處理復(fù)雜一些，但是保證了窗口中數(shù)據(jù)的頻次

另外一種方案是優(yōu)化WATERMARK生成的機(jī)制，如果一段時(shí)間后WATERMARK仍然沒(méi)有變化，那就將WATERMARK自動(dòng)上漲一次到當(dāng)前窗口的結(jié)束時(shí)間，這樣保證窗口處理的延時(shí)有個(gè)上限

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);
    }
}

實(shí)際測(cè)試中發(fā)現(xiàn) WATERMARK是否觸發(fā)和算子的并發(fā)度和WATERMARK生成的位置有關(guān)
測(cè)試結(jié)果如下：

Env default parallism 10: Source parallism 20, window parallism 6, watermark 生成定義在keyby 之前
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為10， WINDOW SUBTASK 并發(fā)度為6, 窗口可以正常觸發(fā)
Env default parallism 20, Source parallism 20, window parallism 6, watermark 生成定義在keyby 之前
Source 到 WINDOW 算子之前合成一個(gè)SUBTASK，并發(fā)度為20， WINDOW SUBTASK 并發(fā)度為6, 窗口可以正常觸發(fā)
Env default parallism 60, Source parallism 20, window parallism 10, watermark 生成定義在keyby 之前
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為60，WINDOW SUBTASK 并發(fā)度為10, 窗口不能正常觸發(fā)(個(gè)人理解原因是算子并發(fā)度擴(kuò)大，導(dǎo)致一些CHANNEL處理線程沒(méi)有數(shù)據(jù)，根據(jù)上文的解釋?zhuān)琖ATERMARK對(duì)齊會(huì)取所有CHANNEL最小的WATERMARK，導(dǎo)致水位無(wú)法上漲
可以從FLINK CONSOLE的WATERMARKS看出)
Env default parallism 60, Source parallism 20, window parallism 10, watermark 生成定義在Source之后
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為60，WINDOW SUBTASK 并發(fā)度為10, 窗口可以正常觸發(fā)
Env default parallism 10, Source parallism 20, window parallism 20, watermark 生成定義在keyby 之前
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為10， WINDOW SUBTASK 并發(fā)度為20, 窗口可以正常觸發(fā)
Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在keyby 之前
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為30， WINDOW SUBTASK 并發(fā)度為20, 窗口不能正常觸發(fā)
Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在keyby 之前
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為30， WINDOW SUBTASK 并發(fā)度為20, 窗口不能正常觸發(fā)
Env default parallism 30, Source parallism 20, window parallism 20, watermark 生成定義在Source 之后
Source 為單獨(dú)的SUBTASK 并發(fā)度為20, 之后到WINDOW算子之前合成一個(gè)SUBTASK，并發(fā)度為30， WINDOW SUBTASK 并發(fā)度為20, 窗口可以正常觸發(fā)

所以注意WINDOW算子之前最好避免讓下游算子的并發(fā)度超過(guò)上游算子，否則就把WATERMARK的生成盡量放到DAG的前端，這樣WATERMARK可以被傳遞到下游算子

本文題目：Flinkwatermark
文章路徑：http://muchs.cn/article18/gdgodp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供Google、商城網(wǎng)站、品牌網(wǎng)站設(shè)計(jì)、小程序開(kāi)發(fā)、響應(yīng)式網(wǎng)站、網(wǎng)站建設(shè)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶(hù)投稿、用戶(hù)轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容