两个流之间的 Scala Flink Join 不起作用

Question

我想加入来自 kafka producer 的两个流，但该连接不起作用。我使用 AssignerWithPeriodicWatermark 来定义我的分配器，我尝试使用 3 分钟的窗口连接两个流。但我没有得到任何输出。我打印了这两个流，以确保它们的事件在时间上足够接近。

object Job {

class Assigner extends AssignerWithPeriodicWatermarks[String] {
  // 1 s in ms
  val bound: Long = 1000
  // the maximum observed timestamp
  var maxTs: Long = Long.MinValue

  override def getCurrentWatermark: Watermark = {
    new Watermark(maxTs - bound)
   }

  override def extractTimestamp(r: String, previousTS: Long): Long = {    
    maxTs = Math.max(maxTs,previousTS)    
    previousTS
   }
}

 def main(args: Array[String]): Unit = {

val env = StreamExecutionEnvironment.getExecutionEnvironment//createLocalEnvironment()
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9093")

properties.setProperty("group.id", "test")
val consumerId = new FlinkKafkaConsumer[String]("topic_id", new SimpleStringSchema(), properties)

val streamId = env.addSource(consumerId).assignTimestampsAndWatermarks(new Assigner)

val streamIdParsed=streamId.map{s =>s.parseJson}.map{ value => (value.asJsObject.getFields("id")(0).toString(),value.asJsObject.getFields("m","w")) }


val consumerV = new FlinkKafkaConsumer[String]("topic_invoice", new SimpleStringSchema(), properties)


val streamV = env.addSource(consumerV).assignTimestampsAndWatermarks(new Assigner)

val streamVParsed = streamV.map{s =>s.parseJson}.map{ value => (value.asJsObject.getFields("id")(0).toString(),value.asJsObject.getFields("products")(0).toString().parseJson.asJsObject.getFields("id2", "id3")) }


    streamIdParsed.join(streamVParsed).where(_._1).equalTo(_._1).window(SlidingEventTimeWindows.of(Time.seconds(60),Time.seconds(1))).apply { (e1, e2) => (e1._1,"test") }.print()
} }

Answer 1

可能出错的事情（由你来检查，因为你提供的信息太少而无法缩小范围）

< li >两个主题都没有关于Kafka的事件 < li >两个主题都没有水印进度 < li >您的数据按分钟计算，代码按秒计算

Answer 2

问题是您尚未设置autoWatermark Interval并且您正在使用周期分配器。您需要执行以下操作：

env.getConfig.setAutowatermarkInterval（[someinterval]）

这将解决未生成水印的问题。