我有一个数据流，看起来像这样：

impressionId | id | name | eventType | timestamp

我需要过滤（忽略）类型为“click”的事件，它没有匹配类型为“impression”的“impressionid”（因此基本上忽略没有impression的clicks事件），然后计算我总共有多少个impression，以及在特定时间窗口中我有多少次单击（对于id/name对）。
我就是这样找到解决方案的：

[...]
Table eventsTable = tEnv.fromDataStream(eventStreamWithTimeStamp, "impressionId, id, name, eventType, eventTime.rowtime");
tEnv.registerTable("Events", eventsTable);
Table clicksTable = eventsTable
      .where("eventType = 'click'")
      .window(Slide.over("24.hour").every("1.minute").on("eventTime").as("minuteWindow"))
      .groupBy("impressionId, id, name, eventType, minuteWindow")
      .select("impressionId as clickImpressionId, eventType as clickEventType, concat(concat(id,'_'), name) as concatClickId, id as clickId, name as clickName, minuteWindow.rowtime as clickMinute");
Table impressionsTable = eventsTable
      .where("eventType = 'impression'")
      .window(Slide.over("24.hour").every("1.minute").on("eventTime").as("minuteWindow"))
      .groupBy("impressionId, id, name, eventType, minuteWindow")
      .select("impressionId as impressionImpressionId, eventType as impressionEventType, concat(concat(id,'_'), name) as concatImpId, id as impId, name as impName, minuteWindow.rowtime as impMinute");
Table filteredClickCount = clicksTable
      .join(impressionsTable, "clickImpressionId = impressionImpressionId && concatClickId = concatImpId && clickMinute = impMinute")
      .window(Slide.over("24.hour").every("1.minute").on("clickMinute").as("minuteWindow"))
      .groupBy("concatClickId, clickMinute")
      .select("concatClickId, concatClickId.count as clickCount, clickMinute as eventTime");
 DataStream<Test3> result = tEnv.toAppendStream(filteredClickCount, Test3.class);
result.print();

我要做的只是创建两个表，一个有点击，一个有印象，'内部'连接点击到印象和一个被连接意味着他们的点击有一个匹配的印象。
现在这不管用，我也不知道为什么！？
最后一个联接表生成的计数不正确。它在第一分钟起作用，但在那之后计数几乎减少了一倍。
然后，我尝试修改最后一个表，如下所示：

Table clickWithMatchingImpression2 = clicksTable
      .join(impressionsTable, "clickImpressionId = impressionImpressionId && concatClickId = concatImpId && clickMinute = impMinute")
      .groupBy("concatClickId, clickMinute")
      .select("concatClickId, concatClickId.count as clickCount, clickMinute as eventTime");
DataStream<Tuple3<Boolean, Tuple3>> result2 = tEnv.toRetractStream(clickWithMatchingImpression2, Test3.class);
    result2.print();

还有。。。。这管用！？但是我不知道为什么，也不知道如何处理这个数据流<tuple3<boolean，test3>>格式。。。当table没有Windows时，Flink拒绝使用toappendstream。我想要一个只有最后数字的简单结构。
1）我的方法正确吗？有没有更简单的方法过滤没有印象的点击？
2）为什么我的解决方案中的计数不正确？