如何在Flink的兩個不同的Kafka流中應用相同的模式？

Question

我具有低於此弗林克程序：

object WindowedWordCount { 
    val configFactory = ConfigFactory.load() 

    def main(args: Array[String]) = { 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 

    val kafkaStream1 = env.addSource(new FlinkKafkaConsumer010[String](topic1, new SimpleStringSchema(), props)) 
     .assignTimestampsAndWatermarks(new TimestampExtractor) 

    val kafkaStream2 = env.addSource(new FlinkKafkaConsumer010[String](topic2, new SimpleStringSchema(), props)) 
     .assignTimestampsAndWatermarks(new TimestampExtractor) 

    val partitionedStream1 = kafkaStream1.keyBy(jsonString => { 
     extractUserId(jsonString) 
    }) 

    val partitionedStream2 = kafkaStream2.keyBy(jsonString => { 
     extractUserId(jsonString) 
    }) 

    //Is there a way to match the userId from partitionedStream1 and partitionedStream2 in this same pattern? 
    val patternForMatchingUserId = Pattern.begin[String]("start") 
     .where(stream1.getUserId() == stream2.getUserId()) //I want to do something like this 

    //Is there a way to pass in partitionedStream1 and partitionedStream2 to this CEP.pattern function? 
    val patternStream = CEP.pattern(partitionedStream1, patternForMatchingUserId) 

    env.execute() 
    } 
} 

在上面的程序弗林克，我有一個名爲partitionedStream1兩個流和partitionedStream2是keyedBy的用戶ID。

我想以某種方式比較patternForMatchingUserId模式中兩個數據流的數據（類似於上面顯示的）。有沒有辦法將兩個流傳遞給CEP.Pattern函數？

事情是這樣的：

val patternStream = CEP.pattern(partitionedStream1, partitionedStream2, patternForMatchingUserId)

Answer 1

沒有辦法，你可以通過兩個流至CEP，但你可以通過組合流。

如果兩個流具有相同的類型/架構。你可以把它們結合起來。我相信這個解決方案符合你的情況。

partitionedStream1.union(partitionedStream2).keyBy(...) 

如果他們有不同的模式。您可以使用一些自定義邏輯將它們轉換爲一個流，例如coFlatMap。