Apache Flink:未触发流加入窗口

Question

>我正在尝试在 apache flink 中加入两个流以获得一些结果。

我的项目的当前状态是，我正在获取Twitter数据并将其映射到一个2元组，其中用户的语言和定义的时间窗口中的推文总和被保存。 我这样做是为了每种语言的推文数量和每种语言的转发次数。推文/转推聚合在其他进程中工作正常。

我现在想获取转推次数占时间窗口中所有推文数量的百分比。

因此，我使用以下代码：

Time windowSize = Time.seconds(15);
// Sum up tweets per language
DataStream<Tuple2<String, Integer>> tweetsLangSum = tweets
.flatMap(new TweetLangFlatMap())
.keyBy(0)
.timeWindow(windowSize)
.sum(1);
// ---
// Get retweets out of all tweets per language
DataStream<Tuple2<String, Integer>> retweetsLangMap = tweets
.keyBy(new KeyByTweetPostId())
.flatMap(new RetweetLangFlatMap());
// Sum up retweets per language
DataStream<Tuple2<String, Integer>> retweetsLangSum = retweetsLangMap
.keyBy(0)
.timeWindow(windowSize)
.sum(1);
// ---
tweetsLangSum.join(retweetsLangSum)
.where(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> tweet) throws Exception {
return tweet.f0;
}
})
.equalTo(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> tweet) throws Exception {
return tweet.f0;
}
})
.window(TumblingEventTimeWindows.of(windowSize))
.apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple4<String, Integer, Integer, Double>>() {
@Override
public Tuple4<String, Integer, Integer, Double> join(Tuple2<String, Integer> in1, Tuple2<String, Integer> in2) throws Exception {
String lang = in1.f0;
Double percentage = (double) in1.f1 / in2.f1;
return new Tuple4<>(in1.f0, in1.f1, in2.f1, percentage);
}
})
.print();

当我打印tweetsLangSum或retweetsLangSum时，输出似乎很好。我的问题是我从来没有从连接中获得输出。有人知道为什么吗？还是我在聚合的第一步中使用窗口函数在连接方面是错误的？

Answer 1

这可能是由不同时间语义的混合引起的。KeyedStream.timeWindow()方法是一种快捷方式，它基于配置的时间特征创建窗口运算符，即，如果启用了事件时间，则为事件时间窗口，否则为处理时间窗口。对于联接，您可以显式定义事件时间窗口。

是否启用了事件时间处理？

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);