Flink 从0到1学习—— Flink 不可以连续 Spli

作者: zhisheng_blog | 来源:发表于2019-07-24 22:29 被阅读0次

前言

今天上午被 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 也可以满足需求，但是就是代码有点啰嗦。