Flink 从 0 到 1 学习—— 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();

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