फ्लिंक में प्रोसेसिंग स्ट्रीम

Question

मुझे संदेशों को संसाधित करने की आवश्यकता है:

1. प्रत्येक संदेश को सभी प्रक्रियाओं से अपरिवर्तित होना चाहिए।
2. यदि प्रक्रिया एक पैटर्न मिलान का पता लगाती है, तो संदेश की एक प्रति दूसरे थ्रेड को पास करनी होगी।

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

मैं 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);

उदाहरण के लिए, लैम्ब्डा 1, 2, आदि हालत जाँच कार्य हैं

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

लेकिन किसी कारण से यह मेरे लिए काम नहीं करता है, शायद कुछ अन्य तरीके हैं? जैसा कि मैं प्रलेखन से समझता हूं (https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html), OutputTag का उपयोग टैग के साथ चिह्नित अतिरिक्त संदेश बनाने के लिए किया जाता है। या मैं गलत हूँ?

Answer 1

जिस तरह से आपने अपने नौकरी के ग्राफ को तार-तार कर दिया है, इसका मतलब है कि नौकरी के अंत की ओर जहां आप साइड आउटपुट का उपयोग करते हैं

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

साइड आउटपुट पर लास्ट प्रोसेस फंक्शन में जो भी हो, वह आपको मिल रहा है - यानी, केवल ईवेंट्स lambda7 द्वारा।

मेरा मानना ​​है कि आप ऐसा करने का इरादा रखते हैं

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

इसके अलावा, मुझे विश्वास नहीं है कि आप का उपयोग किया जाना चाहिए startNewChain()(जो ऑपरेटर चाइनिंग को अक्षम करता है)। ऑपरेटर चेनिंग एक मूल्यवान अनुकूलन है, और इसे केवल विशेष परिस्थितियों में अक्षम किया जाना चाहिए।