Flink间隔连接不输出

Question

我们有一个Flink任务，它将两个流连接起来，两个流都使用来自Kafka的事件。下面是示例代码

val articleEventStream: DataStream[ArticleEvent] = env.addSource(articleEventSource)
  .assignTimestampsAndWatermarks(new ArticleEventAssigner) 
val feedbackEventStream: DataStream[FeedbackEvent] = env.addSource(feedbackEventSource)
  .assignTimestampsAndWatermarks(new FeedbackEventAssigner) 

articleEventStream
    .keyBy(article => article.id)
    .intervalJoin(feedbackEventStream.keyBy(feedback => feedback.article.id))
    .between(Time.seconds(-5), Time.seconds(10))
    .process(new ProcessJoinFunction[ArticleEvent, FeedbackEvent, String] {
        override def processElement(left: ArticleEvent, right: FeedbackEvent, ctx: ProcessJoinFunction[ArticleEvent, FeedbackEvent, String]#Context, out: Collector[String]): Unit = {
         out.collect(left.name + " got feedback: " + right.feedback); 
        }
      });
});

class ArticleEventAssigner extends AssignerWithPunctuatedWatermarks[ArticleEvent] {
  val bound: Long = 5 * 1000

  override def checkAndGetNextWatermark(lastElement: ArticleEvent, extractedTimestamp: Long): Watermark = {
    new Watermark(extractedTimestamp - bound)
  }

  override def extractTimestamp(element: ArticleEvent, previousElementTimestamp: Long): Long = {
    element.occurredAt
  }
}

class FeedbackEventAssigner extends AssignerWithPunctuatedWatermarks[FeedbackEvent] {
  val bound: Long = 5 * 1000

  override def checkAndGetNextWatermark(lastElement: FeedbackEvent, extractedTimestamp: Long): Watermark = {
    new Watermark(extractedTimestamp - bound)
  }

  override def extractTimestamp(element: FeedbackEvent, previousElementTimestamp: Long): Long = {
    element.occurredAt
  }
}

但是，我们没有看到任何连接输出。我们检查了每个流是否连续发射带有时间戳和适当水印的元素。有人知道可能的原因吗？

Answer 1

在检查了不同的部分（时间戳/水印、触发器）之后，我注意到我犯了一个错误，即我使用的窗口大小

之间（Time.seconds（-5），Time.seconds（10））

只是太小了，无法从两个流中找到要连接的元素。这听起来可能很明显，但是因为我是Flink的新手，我不知道去哪里检查。所以，我的教训是，如果连接不输出，可能有必要检查窗口大小。感谢所有的评论！