Flink CEP没有结果打印

我想打印出一个字符串，如果你好和世界都使用Flink CEP库找到。我的来源是Kafka并使用控制台生产者输入数据。这部分工作。我可以打印出我在主题中输入的内容。但是，它不会打印出我最终的信息“世界真好！”。它甚至不会打印出它输入的lambda。以下是该类Flink CEP没有结果打印

package kafka; 

import org.apache.flink.cep.CEP; 
import org.apache.flink.cep.PatternStream; 
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.FlinkKafkaConsumer08; 
import org.apache.flink.streaming.util.serialization.SimpleStringSchema; 
import org.apache.flink.util.Collector; 

import java.util.Map; 
import java.util.Properties; 

/** 
* Created by crackerman on 9/16/16. 
*/ 
public class WordCount { 

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

    Properties properties = new Properties(); 
    properties.put("bootstrap.servers", "localhost:9092"); 
    properties.put("zookeeper.connect", "localhost:2181"); 
    properties.put("group.id", "test"); 
    StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment(); 
    see.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 

    DataStream<String> src = see.addSource(new FlinkKafkaConsumer08<>("complexString", 
                     new SimpleStringSchema(), 
                     properties)); 

    src.print(); 


    Pattern<String, String> pattern = Pattern.<String>begin("first") 
      .where(evt -> evt.contains("Hello")) 
      .followedBy("second") 
      .where(evt -> evt.contains("World")); 

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

    DataStream<String> alerts = patternStream.flatSelect(
      (Map<String, String> in, Collector<String> out) -> { 
       System.out.println("Made it to the lambda"); 
       String first = in.get("first"); 
       String second = in.get("second"); 
       System.out.println("First: " + first); 
       System.out.println("Second: " + second); 

       if (first.equals("Hello") && second.equals("World")) { 

        out.collect("The world is so nice!"); 
       } 


      }); 

    alerts.print(); 

    see.execute(); 
} 

}

任何帮助将不胜感激。

谢谢！

来源

2016-09-19 Crackerman

的问题是以下行

see.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

如果被删除，它的工作原理我预期的方式。

来源

2016-09-19 16:08:02 Crackerman

您是否指定了时间戳和水印？如果您告诉Flink每条记录的时间和已经过了多长时间，事件时间模式才起作用。没有这些信息，Flink无法在事件时间处理数据。如果您的数据无序到达或者您有基于时间的操作（例如窗口），则需要事件时间才能产生一致的结果。 –

我没有分配时间戳和水印。我已经回去并实施了这个。谢谢（你的）信息！ – Crackerman

Flink CEP没有结果打印

回答

相关问题