Почему одни и те же события PatternStream могут быть отправлены как в PatternSelectFunction, так и в PatternTimeoutFunction?

Question

Почему одни и те же события PatternStream могут быть отправлены как в PatternSelectFunction, так и в PatternTimeoutFunction?

Я должен собрать 3 события в 3 потоках из Кафки, имеющих одинаковый корреляционный идентификатор в данный момент времени, и быть в состоянии собрать все или часть этих событий, если они прибудут поздно.

Я использовал объединение на 3 DataStream и шаблон CEP. Но я замечаю, что события, которые хорошо совпадают с шаблоном и, следовательно, собираются в функции выбора, также отправляются в функции тайм-аута после достижения тайм-аута.

Я не знаю, что я сделал неправильно в своем примере, или что я не понял, но я ожидал, что события, которые положительно совпадают, также не имеют времени ожидания.

У меня складывается впечатление, что несвязанные моментальные снимки сохраняются.

Я использую версию 1.3.0 Flink.

Спасибо за помощь.

Вывод на консоль, где мы видим, что 2 из 3 коррелированных событий выбраны и рассчитаны по времени:

Соответствующие события:
Key --- 0b3c116e-0703-43cb-8b3e-54b0b5e93948
Key --- f969dd4d-47ff-445c-9182-0f95a569febb
Key --- 2ecbb89d-1463-4669-a657-555f73b6fb1d

Тайм-аут события:

Первый вызов функции тайм-аута:
Key --- f969dd4d-47ff-445c-9182-0f95a569febb
Key --- 0b3c116e-0703-43cb-8b3e-54b0b5e93948

Второй звонок:
Key --- f969dd4d-47ff-445c-9182-0f95a569febb

11:01:44,677 INFO  com.bnpp.pe.cep.Main                                          - Matching events:
11:01:44,678 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep2Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---0b3c116e-0703-43cb-8b3e-54b0b5e93948, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
11:01:44,678 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep1Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---2ecbb89d-1463-4669-a657-555f73b6fb1d, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
11:01:44,678 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep3Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---f969dd4d-47ff-445c-9182-0f95a569febb, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
Right(SctRequestFinalEvent(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---2196fdb0-01e8-4cc6-af4b-04bcf9dc67a2, debtorIban=null, creditorIban=null, amount=null, communication=null), state=SUCCESS))
11:01:49,635 INFO  com.bnpp.pe.cep.Main                                          - Timed out events:
11:01:49,636 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep3Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---f969dd4d-47ff-445c-9182-0f95a569febb, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
11:01:49,636 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep2Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---0b3c116e-0703-43cb-8b3e-54b0b5e93948, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
11:01:49,636 INFO  com.bnpp.pe.cep.Main                                          - Timed out events:
11:01:49,636 INFO  com.bnpp.pe.cep.Main                                          - SctRequestProcessStep3Event(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---f969dd4d-47ff-445c-9182-0f95a569febb, debtorIban=BE42063929068055, creditorIban=BE42063929068056, amount=100.0, communication=test), succeeded=false)
Left(SctRequestFinalEvent(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---aa437bcf-ecaa-4561-9f4e-08a902f0e248, debtorIban=null, creditorIban=null, amount=null, communication=null), state=FAILED))
Left(SctRequestFinalEvent(super=SctRequestEvent(correlationId=cId---a14a4e23-56c5-4242-9c43-d465d2b84454, key=Key---5420eb41-2723-42ac-83fd-d203d6bf2526, debtorIban=null, creditorIban=null, amount=null, communication=null), state=FAILED))

Мой тестовый код:

package com.bnpp.pe.cep;

import com.bnpp.pe.event.Event;
import com.bnpp.pe.event.SctRequestFinalEvent;
import com.bnpp.pe.util.EventHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by Laurent Bauchau on 2/08/2017.
 */
@Slf4j
public class Main implements Serializable {

    public static void main(String... args) {
        new Main();
    }

    public static final String step1Topic = "sctinst-step1";
    public static final String step2Topic = "sctinst-step2";
    public static final String step3Topic = "sctinst-step3";

    private static final String PATTERN_NAME = "the_3_correlated_events_pattern";

    private final FlinkKafkaConsumer010<Event> kafkaSource1;
    private final DeserializationSchema<Event> deserializationSchema1;

    private final FlinkKafkaConsumer010<Event> kafkaSource2;
    private final DeserializationSchema<Event> deserializationSchema2;

    private final FlinkKafkaConsumer010<Event> kafkaSource3;
    private final DeserializationSchema<Event> deserializationSchema3;

    private Main() {

        // Kafka init
        Properties kafkaProperties = new Properties();
        kafkaProperties.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProperties.setProperty("zookeeper.connect", "localhost:2180");
        kafkaProperties.setProperty("group.id", "sct-validation-cgroup1");

        deserializationSchema1 = new SctRequestProcessStep1EventDeserializer();
        kafkaSource1 = new FlinkKafkaConsumer010<>(step1Topic, deserializationSchema1, kafkaProperties);

        deserializationSchema2 = new SctRequestProcessStep2EventDeserializer();
        kafkaSource2 = new FlinkKafkaConsumer010<>(step2Topic, deserializationSchema2, kafkaProperties);

        deserializationSchema3 = new SctRequestProcessStep3EventDeserializer();
        kafkaSource3 = new FlinkKafkaConsumer010<>(step3Topic, deserializationSchema3, kafkaProperties);

        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

            DataStream<Event> s1 = env.addSource(kafkaSource1);
            DataStream<Event> s2 = env.addSource(kafkaSource2);
            DataStream<Event> s3 = env.addSource(kafkaSource3);

            DataStream<Event> unionStream = s1.union(s2, s3);

            Pattern successPattern = Pattern.<Event>begin(PATTERN_NAME)
                    .times(3)
                    .within(Time.seconds(5));

            PatternStream<Event> matchingStream = CEP.pattern(
                    unionStream.keyBy(new CIDKeySelector()),
                    successPattern);

            matchingStream.select(new MyPatternTimeoutFunction(), new MyPatternSelectFunction())
                    .print()
                    .setParallelism(1);

            env.execute();

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private static class MyPatternTimeoutFunction implements PatternTimeoutFunction<Event, SctRequestFinalEvent> {

        @Override
        public SctRequestFinalEvent timeout(Map<String, List<Event>> pattern, long timeoutTimestamp) throws Exception {

            List<Event> events = pattern.get(PATTERN_NAME);
            log.info("Timed out events:");
            events.forEach(e -> log.info(e.toString()));

            // Resulting event creation
            SctRequestFinalEvent event = new SctRequestFinalEvent();
            EventHelper.correlate(events.get(0), event);
            EventHelper.injectKey(event);
            event.setState(SctRequestFinalEvent.State.FAILED);

            return event;
        }
    }

    private static class MyPatternSelectFunction
            implements PatternSelectFunction<Event, SctRequestFinalEvent> {

        @Override
        public SctRequestFinalEvent select(Map<String, List<Event>> pattern) throws Exception {

            List<Event> events = pattern.get(PATTERN_NAME);
            log.info("Matching events:");
            events.forEach(e -> log.info(e.toString()));

            // Resulting event creation
            SctRequestFinalEvent event = new SctRequestFinalEvent();
            EventHelper.correlate(events.get(0), event);
            EventHelper.injectKey(event);
            event.setState(SctRequestFinalEvent.State.SUCCESS);

            return event;
        }
    }

    private static class CIDKeySelector implements KeySelector<Event, String> {
        @Override
        public String getKey(Event event) throws Exception {
            return event.getCorrelationId();
        }
    }
}

0

apache-flink flink-cep

Источник

user2683199 07 авг '17 в 09:22

2 ответа

Решение

Ваша программа....

В вашей программе выделите текст по времени, поэтому вы передаете объект PatterStream ОБА функции. Вам не нужно время для выбора строки... Вы не переходите к PatternTimeOutFunction().

Смотрите здесь, без фактора времени.

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Map;

public class FlinkCEP {

    public static void main(String[] args) throws Exception {

        // set up the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> text = env.socketTextStream("localhost", 1111)
                .flatMap(new LineTokenizer());

        text.print();

        Pattern<String, String> pattern =
                Pattern.<String>begin("start").where(txt -> txt.equals("a"))
                       .next("middle").where(txt -> txt.equals("b"))
                       .followedBy("end").where(txt -> txt.equals("c")).within(Time.seconds(1));

        PatternStream<String> patternStream = CEP.pattern(text, pattern);

        DataStream<String> alerts = patternStream.select(new PatternSelectFunction<String, String>() {
            @Override
            public String select(Map<String, String> matches) throws Exception {
                return "Found: " +
                        matches.get("start") + "->" +
                        matches.get("middle") + "->" +
                        matches.get("end");
            }
        });

        // emit result
        alerts.print();

        // execute program
        env.execute("WordCount Example");
    }
}

0

Источник

user8209028 07 авг '17 в 11:49

Другие вопросы по тегам apache-flink flink-cep

user4250114 07 авг '17 в 11:47 2017-08-07 11:47 · Accepted Answer · 2017-08-07 11:47

Давайте проанализируем, что говорит ваш паттерн. Вы передаете образец как:

Pattern.<Event>begin(PATTERN_NAME)
    .times(3)
    .within(Time.seconds(5));

В нем говорится, искать последовательность из трех любых событий, которые происходят в течение 5 секунд. Прямо сейчас Flink начинает поиск нового соответствия каждому последующему событию (продолжается работа по внедрению нового MatchingBehavioursсм. ФЛИНК-7169).

Итак, чтобы показать простой пример. Если у вас есть последовательность, как A B C D E в течение 5 секунд. Библиотека CEP вернет результаты:

азбука
BCD
CDE

и два раза в день:

Делавэр
D