Flux de traitement dans Flink

Question

Je dois traiter les messages comme suit:

1. chaque message doit passer inchangé à travers tous les processus.
2. si le processus détecte une correspondance de modèle, une copie du message doit être transmise à un autre thread.

streamIn -> process1 -> process2 -> .... -> processN
                 \         |                /
                  \        |               /
                     sink to kafka

J'ai essayé d'utiliser OutputTag

 OutputTag<SysmonPartial> filtersOutput = new OutputTag<SysmonPartial>("FiltersOutput"){};

 DataStream<SysmonPartial> kafkaSource = env.addSource(consumer);

        DataStream<SysmonPartial> source = kafkaSource.rebalance();

        SingleOutputStreamOperator<SysmonPartial> s = source
                .process(lambda1).name("lambda1").startNewChain()
                .process(lambda2).name("lambda2").startNewChain()
                .process(lambda3).startNewChain()
                .process(lambda4).startNewChain()
                .process(lambda5).startNewChain()
                .process(lambda6).startNewChain()
                .process(lambda7).startNewChain();

        SingleOutputStreamOperator<Any> output = s
                .getSideOutput(filtersOutput)
                .process(filterProcessFunction).setParallelism(1).startNewChain();

        output.addSink(sink).setParallelism(1);

        env.execute(jobName);

où lambda 1, 2, etc. sont des fonctions de vérification de condition, par exemple

public class Bypass_WS_01_03 extends FilterTagFuction<SysmonPartial, SysmonPartial, SysmonPartial> {
    private static final Pattern p_1 = Pattern.compile("pattern1");
    private static final Pattern p_0 = Pattern.compile("pattern2");

    @Override
    public void processElement(SysmonPartial t, Context ctx, Collector<SysmonPartial> out) throws Exception {
        out.collect(t);
        if (
                "1".equals(t.B_VendorEventID) &&
                        t.CommandLine != null && t.CommandLine.length() != 0 && p_0.matcher(t.CommandLine).find() &&
                        t.ImageName != null && t.ImageName.length() != 0 && p_1.matcher(t.ImageName).find()
        ) {
            t.RuleId = "Bypass_WS_01_03";
            ctx.output(getOutputTag(), t);
        }
    }
}

Mais pour une raison quelconque, cela ne fonctionne pas pour moi, peut-être y a-t-il d'autres méthodes? D'après ce que je comprends de la documentation (https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html), OutputTag est utilisé pour créer des messages supplémentaires marqués avec Tag. Ou ai-je tort?

Answer 1

La façon dont vous avez câblé votre graphique de tâche signifie que vers la fin de la tâche, vous accédez à la sortie latérale

SingleOutputStreamOperator<Any> output = s
    .getSideOutput(filtersOutput)

vous n'obtenez que la dernière fonction de processus mise sur la sortie latérale, c'est-à-dire uniquement les événements émis par lambda7.

Ce que je pense que vous avez l'intention de faire peut être exprimé comme

SingleOutputStreamOperator<...> s1 = source.process(lambda1).name("lambda1");
SingleOutputStreamOperator<...> s2 = s1.process(lambda2).name("lambda2");
SingleOutputStreamOperator<...> s3 = s2.process(lambda3);
...

DataStream<...> side1 = s1.getSideOutput(filtersOutput);
DataStream<...> side2 = s2.getSideOutput(filtersOutput);
DataStream<...> side3 = s3.getSideOutput(filtersOutput);
...

SingleOutputStreamOperator<Any> output = side1.union(side2, side3, ...)
    .process(filterProcessFunction)
    ...

De plus, je ne pense pas que vous devriez utiliser startNewChain()(ce qui désactive le chaînage des opérateurs). Le chaînage des opérateurs est une optimisation précieuse et ne doit être désactivé que dans des circonstances spéciales.