Flink 從0到1學習—— Flink 不可以連續 Split(分流)？

前言今天上午被 Flink 的一個運算元困惑了下，具體問題是什麼呢？我有這麼個需求：有不同種類型的告警數據流(包含恢複數據)，然後我要將這些數據流做一個拆分，拆分後的話，每種告警裡面的數據又想將告警數據和恢複數據拆分出來。結果，這個需求用 Flink 的 Split 運算符出現了問題。分析需 ...

前言

今天上午被 Flink 的一個運算元困惑了下，具體問題是什麼呢？

我有這麼個需求：有不同種類型的告警數據流(包含恢複數據)，然後我要將這些數據流做一個拆分，拆分後的話，每種告警裡面的數據又想將告警數據和恢複數據拆分出來。

結果，這個需求用 Flink 的 Split 運算符出現了問題。

分析

需求如下圖所示：

我是期望如上這樣將數據流進行拆分的，最後將每種告警和恢復用不同的消息模版做一個渲染，渲染後再通過各種其他的方式（釘釘群
郵件、簡訊）進行告警通知。

於是我的代碼大概的結構如下代碼所示：

//dataStream 是總的數據流

//split 是拆分後的數據流
SplitStream<AlertEvent> split = dataStream.split(new OutputSelector<AlertEvent>() {
    @Override
    public Iterable<String> select(AlertEvent value) {
        List<String> tags = new ArrayList<>();
        switch (value.getType()) {
            case MIDDLEWARE:
                tags.add(MIDDLEWARE);
                break;
            case HEALTH_CHECK:
                tags.add(HEALTH_CHECK);
                break;
            case DOCKER:
                tags.add(DOCKER);
                break;
            //...
            //當然這裡還可以很多種類型
        }
        return tags;
    }
});

//然後你想獲取每種不同的數據類型，你可以使用 select
DataStream<AlertEvent> middleware = split.select(MIDDLEWARE);   //選出中間件的數據流

//然後你又要將中間件的數據流分流成告警和恢復
SplitStream<AlertEvent> middlewareSplit = middleware.split(new OutputSelector<AlertEvent>() {
    @Override
    public Iterable<String> select(AlertEvent value) {
        List<String> tags = new ArrayList<>();
        if(value.isRecover()) {
            tags.add(RECOVER)
        } else {
            tags.add(ALERT)
        }
        return tags;
    }
});
middlewareSplit.select(ALERT).print();    
        


DataStream<AlertEvent> healthCheck = split.select(HEALTH_CHECK);   //選出健康檢查的數據流

//然後你又要將健康檢查的數據流分流成告警和恢復
SplitStream<AlertEvent> healthCheckSplit = healthCheck.split(new OutputSelector<AlertEvent>() {
    @Override
    public Iterable<String> select(AlertEvent value) {
        List<String> tags = new ArrayList<>();
        if(value.isRecover()) {
            tags.add(RECOVER)
        } else {
            tags.add(ALERT)
        }
        return tags;
    }
});
healthCheckSplit.select(ALERT).print();



DataStream<AlertEvent> docekr = split.select(DOCKER);   //選出容器的數據流

//然後你又要將容器的數據流分流成告警和恢復
SplitStream<AlertEvent> dockerSplit = docekr.split(new OutputSelector<AlertEvent>() {
    @Override
    public Iterable<String> select(AlertEvent value) {
        List<String> tags = new ArrayList<>();
        if(value.isRecover()) {
            tags.add(RECOVER)
        } else {
            tags.add(ALERT)
        }
        return tags;
    }
});
dockerSplit.select(ALERT).print();

結構我抽象後大概就長上面這樣，然後我先本地測試的時候只把容器的數據那塊代碼打開了，其他種告警的分流代碼註釋掉了，一運行，發現竟然容器告警的數據怎麼還摻雜著健康檢查的數據也一起列印出來了，一開始我以為自己出了啥問題，就再起碼運行了三遍 IDEA 才發現結果一直都是這樣的。

於是，我只好在第二步分流前將 docekr 數據流列印出來，發現是沒什麼問題，列印出來的數據都是容器相關的，沒有摻雜著其他種的數據啊。這會兒遍陷入了沉思，懵逼發呆了一會。

解決問題

於是還是開始面向 Google 編程：

發現第一條就找到答案了，簡直不要太快，點進去可以看到他也有這樣的需求：

然後這個小伙伴還掙扎了下用不同的方法（雖然結果更慘）：

最後換了個姿勢就好了（果然小伙子會的姿勢挺多的）：

但從這篇文章中，我找到了關聯到的兩個 Flink Issue，分別是：

1、https://issues.apache.org/jira/browse/FLINK-5031

2、https://issues.apache.org/jira/browse/FLINK-11084

然後呢，從第二個 Issue 的討論中我發現了一些很有趣的討論：

對話很有趣，但是我突然想到之前我的知識星球裡面一位很細心的小伙伴問的一個問題了：

可以發現代碼上確實是標明瞭過期了，但是註釋裡面沒寫清楚推薦用啥，幸好我看到了這個 Issue，不然腦子裡面估計這個問題一直會存著呢。

那麼這個問題解決方法是不是意味著就可以利用 Side Outputs 來解決呢？當然可以啦，官方都推薦了，還不能都話，那麼不是打臉啪啪啪的響嗎？不過這裡還是賣個關子將 Side Outputs 後面專門用一篇文章來講，感興趣的可以先看看官網介紹：https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html

另外其實也可以通過 split + filter 組合來解決這個問題，反正關鍵就是不要連續的用 split 來分流。

用 split + filter 的方案代碼大概如下：

DataStream<AlertEvent> docekr = split.select(DOCKER);   //選出容器的數據流

//容器告警的數據流
docekr.filter(new FilterFunction<AlertEvent>() {
    @Override
    public boolean filter(AlertEvent value) throws Exception {
        return !value.isRecover();
    }
})
.print();
        
//容器恢復的數據流        
docekr.filter(new FilterFunction<AlertEvent>() {
    @Override
    public boolean filter(AlertEvent value) throws Exception {
        return value.isRecover();
    }
})
.print();

上面這種就是多次 filter 也可以滿足需求，但是就是代碼有點啰嗦。