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沒有結果打印

回答

相關問題