在Apache Flink中处理不生成警报的事件

Question

我正在使用Flink CEP，我需要处理甚至不生成警报的事件。请问我该怎么做？

我正在使用rabbitMq中的事件，并定义了一些模式。现在我需要做的是将另一个队列中接收到的所有事件发送到一个远程API。我是Flink的新手，所以我遵循了文档中的示例。当我在将接收到的事件与定义的模式进行匹配后尝试发送它们时，我只会得到与模式匹配的结果。例如，我想做的就是在我的事件中将一个属性设置为true，然后将它们全部发送到输出队列。

    public static void cep() throws Exception {
    /**
     * RabbitMQ connection
     */
    final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
            .setHost(HOST)
            .setPort(PORTS[RD.getValue()])
            .setUserName("guest")
            .setPassword("guest")
            .setVirtualHost("/")
            .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    /**
     * Retrieve data inputEventstream from rabbitMQ
     */
    final DataStream<String> inputEventstream = env
            .addSource(new RMQSource<>(
                    connectionConfig, // config for the RabbitMQ connection
                    "input", // name of the RabbitMQ queue to consume
                    true, // use correlation ids; can be false if only at-least-once is required
                    new SimpleStringSchema())) // deserialization schema to turn messages into Java objects
            .setParallelism(1);
    /**
     * Change DataStream<String> to DataStream<MonitoringEvent> where
     * MonitoringEvent refer to a class which modelize our event.
     */
    DataStream<MonitoringEvent> inputEventStreamClean = inputEventstream.flatMap(new Tokenizer());

    Pattern<MonitoringEvent, ?> warningPattern = Pattern.<MonitoringEvent>begin("start")
            .subtype(MonitoringEvent.class)
            .where(new SimpleCondition<MonitoringEvent>() {
                @Override
                public boolean filter(MonitoringEvent value) {
                        return Integer.parseInt(value.getAncienneChute())>=CHUTE_GRAVE;
                }
            }).or(new SimpleCondition<MonitoringEvent>() {
                @Override
                public boolean filter(MonitoringEvent value) {
                        return value.isChaiseRoulante();
                }
            }).or(new SimpleCondition<MonitoringEvent>() {
                    @Override
                    public boolean filter(MonitoringEvent value) {
                        return value.isDeambulateur();
                }
            }).or(new SimpleCondition<MonitoringEvent>() {
                @Override
                public boolean filter(MonitoringEvent value) {
                    return value.isDeambulateur();
            }
            })
            .or(new SimpleCondition<MonitoringEvent>() {
                @Override
                public boolean filter(MonitoringEvent value) {
                    return EntityManager.getInstance().hasCurrentYearFallTwice(value.getIdClient());
            }
            });

    //PatternStream<MonitoringEvent> fallPatternStream = CEP.pattern(inputEventStreamClean.keyBy("idClient"), warningPattern);
    inputEventStreamClean.print();

    // Create a pattern stream from our warning pattern
    PatternStream<MonitoringEvent> tempPatternStream = CEP.pattern(
            inputEventStreamClean.keyBy("idClient"),
            warningPattern);

    DataStream<FallWarning> warnings = tempPatternStream.select(
            (Map<String, List<MonitoringEvent>> pattern) -> {
                MonitoringEvent first = (MonitoringEvent) pattern.get("start").get(0);
                return new FallWarning(first.getIdClient(), Integer.valueOf(first.getAncienneChute()));
            }
    );

    // Alert pattern: Two consecutive temperature warnings appearing within a time interval of 20 seconds
    Pattern<FallWarning, ?> alertPattern = Pattern.<FallWarning>begin("start");

    // Create a pattern stream from our alert pattern
    PatternStream<FallWarning> alertPatternStream = CEP.pattern(
            //warnings.keyBy("idClient"),
                warnings,
            alertPattern);

    // Generate alert 
    DataStream<Alert> alerts = alertPatternStream.flatSelect(
            (Map<String, List<FallWarning>> pattern, Collector<Alert> out) -> {
                FallWarning first = pattern.get("start").get(0);

                if (first.idNiveauUrgence>=CHUTE_GRAVE && (first.isChaiseRoulante() || first.isDeambulateur() || first.isFracture())) {
                    out.collect(new Alert(first.idClient));
                }
            });

    // Print the warning and alert events to stdout
    warnings.print();

    alerts.print(); // here I can send them to RabbitMq

    env.execute();
}

Answer 1

您只需要将Sink添加到您的“警报”DataStream中，例如

alert.addSink(new RMQSink<String>(
connectionConfig,            // config for the RabbitMQ connection
"queueName",                 // name of the RabbitMQ queue to send messages to
new SimpleStringSchema()));  // serialization schema to turn Java objects to messages

每个例子在

https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/connectors/rabbitmq.html