Stream in Flink verarbeiten

Question

Ich muss Nachrichten wie folgt verarbeiten:

1. Jede Nachricht muss alle Prozesse unverändert durchlaufen.
2. Wenn der Prozess eine Musterübereinstimmung feststellt, muss eine Kopie der Nachricht an einen anderen Thread übergeben werden.

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

Ich habe versucht, OutputTag zu verwenden

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

wobei Lambda 1, 2 usw. beispielsweise Zustandsüberprüfungsfunktionen sind

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

Aber aus irgendeinem Grund funktioniert es bei mir nicht, vielleicht gibt es noch andere Methoden? Wie ich aus der Dokumentation verstehe (https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html), OutputTag wird verwendet, um zusätzliche Nachrichten zu erstellen, die mit Tag markiert sind. Oder liege ich falsch?

Answer 1

Die Art und Weise, wie Sie Ihr Jobdiagramm verkabelt haben, bedeutet, dass Sie gegen Ende des Jobs auf die Seitenausgabe zugreifen

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

Sie erhalten nur die letzte Prozessfunktion, die auf den Seitenausgang gestellt wurde - dh nur die von lambda7 ausgegebenen Ereignisse.

Was ich glaube, dass Sie vorhaben, kann ausgedrückt werden als

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)
    ...

Ich glaube auch nicht, dass Sie verwenden sollten startNewChain()(wodurch die Verkettung von Operatoren deaktiviert wird). Die Verkettung von Bedienern ist eine wertvolle Optimierung und sollte nur unter besonderen Umständen deaktiviert werden.