没有正在从弗林克图案化流

Question

打印出来我有下面这样的代码：

import java.util.Properties 

import com.google.gson._ 
import com.typesafe.config.ConfigFactory 
import org.apache.flink.cep.scala.pattern.Pattern 
import org.apache.flink.cep.scala.CEP 
import org.apache.flink.streaming.api.TimeCharacteristic 
import org.apache.flink.streaming.api.scala._ 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010 
import org.apache.flink.streaming.util.serialization.SimpleStringSchema 

object WindowedWordCount { 
    val configFactory = ConfigFactory.load() 
    def main(args: Array[String]) = { 
    val brokers = configFactory.getString("kafka.broker") 
    val topicChannel1 = configFactory.getString("kafka.topic1") 

    val props = new Properties() 
    ... 

    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 

    val dataStream = env.addSource(new FlinkKafkaConsumer010[String](topicChannel1, new SimpleStringSchema(), props)) 

    val partitionedInput = dataStream.keyBy(jsonString => { 
     val jsonParser = new JsonParser() 
     val jsonObject = jsonParser.parse(jsonString).getAsJsonObject() 
     jsonObject.get("account") 
    }) 

    val priceCheck = Pattern.begin[String]("start").where({jsonString => 
     val jsonParser = new JsonParser() 
     val jsonObject = jsonParser.parse(jsonString).getAsJsonObject() 
     jsonObject.get("account").toString == "iOS"}) 

    val pattern = CEP.pattern(partitionedInput, priceCheck) 

    val newStream = pattern.select(x => 
     x.get("start").map({str => 
     str 
     }) 
    ) 

    newStream.print() 

    env.execute() 
    } 
} 

出于某种原因在于newStream.print()没有上面的代码被打印出来。我肯定卡夫卡的数据与我在上面定义的模式相匹配，但由于某种原因没有任何内容正在打印出来。

任何人都可以帮我找出这段代码中的错误吗？

编辑：

class TimestampExtractor extends AssignerWithPeriodicWatermarks[String] with Serializable { 

    override def extractTimestamp(e: String, prevElementTimestamp: Long) = { 
    val jsonParser = new JsonParser() 
    val context = jsonParser.parse(e).getAsJsonObject.getAsJsonObject("context") 
    Instant.parse(context.get("serverTimestamp").toString.replaceAll("\"", "")).toEpochMilli 
    } 

    override def getCurrentWatermark(): Watermark = { 
    new Watermark(System.currentTimeMillis()) 
    } 
} 

我的弗林克文件，让您可以在extractTimestamp方法在getCurrentWatermark方法只返回prevElementTimestamp（如果您正在使用Kafka010）和new Watermark(System.currentTimeMillis)看到。

但我不明白什么prevElementTimestamp是或为什么会返回new Watermark(System.currentTimeMillis)作为WaterMark而不是别的。请你详细说明我们为什么要这样做WaterMark和EventTime请一起工作？

Answer 1

你的设置工作在EventTime，但你不提供时间戳和水印提取。

有关在活动时间工作的更多信息，请参阅docs。如果你想使用卡夫卡嵌入时间戳，这docs可能会帮助你。

在EventTime CEP库在水印抵达时缓冲事件，以正确处理乱序事件。在你的情况下，没有水印生成，所以事件缓冲无限。

---

编辑：

1. 为prevElementTimestamp我认为文档是相当清楚的：

   没有必要使用从卡夫卡的时间戳，当定义一个时间戳提取。 extractTimestamp（）方法的previousElementTimestamp参数包含由Kafka消息携带的时间戳。

   由于Kafka 0.10.x卡夫卡消息可以嵌入时间戳。

2. 在这种情况下生成Watermark为new Watermark(System.currentTimeMillis)不是一个好主意。根据您对数据的了解，您应该创建Watermark。关于我如何能Watermark和EventTime一起工作不会比docs