如何通过datastrem API或Flink Table API/SQL将给定键和公共窗口上的三个或多个数据流/表连接起来？

Question

dataStream.join(otherStream)
.where(<key selector>).equalTo(<key selector>)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.apply (new JoinFunction () {...});

=================================

对于FlinK表API和SQL同样的问题，如何在给定的键和公共窗口上连接三个或更多的表？官方文档https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/table/sql.html只给出了下面单表的示例。

Table result1 = tableEnv.sqlQuery(
"SELECT user, " +
"  TUMBLE_START(rowtime, INTERVAL '1' DAY) as wStart,  " +
"  SUM(amount) FROM Orders " +
"GROUP BY TUMBLE(rowtime, INTERVAL '1' DAY), user");

我尝试编写如下所示的SQL，以便在给定的键和公共窗口上连接三个表，但我认为这是不对的。

String SQL = "SELECT" +
            " grades.user1  , SUM(salaries.amount)   FROM grades " +
            " INNER JOIN salaries ON   grades.user1 =   salaries.user1 " +
            " INNER JOIN person ON   grades.user1 =   person.user1 "+
             "GROUP BY grades.user1, TUMBLE(grades.proctime,  INTERVAL '5' SECOND) "   

对于Flink SQL，我所需要的，就像下面的伪代码一样，是用一个通用的TumblingEventTimeWindow连接三个表，也就是说DataStream API的替代版本，不管用Flink SQL表示，也意味着连接发生在同一TumblingEventTimeWindow中的三个表中的所有事件。

SELECT A.a, B.b, C.c
FROM A, B, C
WHERE A.x = B.x AND A.x = C.x AND
window(TumblingEventTimeWindows.of(Time.seconds(3))

在下面的Flink设计文档中似乎也提到了连接特性：“事件时间滚动窗口流-流连接：连接处于同一滚动事件时间窗口中的两个流的元组”，我不知道Flink SQL是否实现了这种类型的Flink SQL连接特性。

Answer 1

很难对您的问题给出一个明确的答案，因为您需要的连接的语义不清楚。数据流API的窗口连接实现的语义不同于表API/SQL的窗口连接。

在DataStream API上，您可以简单地定义另一个联接，如下所示：

firstStream
  .join(secondStream)
    .where(<key selector>).equalTo(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...})
  .join(thirdStream)
    .where(<key selector>).equalTo(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...})

由于Flink实现了标准SQL，您可以像往常一样定义三个表的联接：

SELECT A.a, B.b, C.c
  FROM A, B, C
  WHERE A.x = B.x AND A.x = C.x AND
        A.ts BETWEEN B.ts - INTERVAL '10' MINUTE AND B.ts + INTERVAL '10' MINUTE AND
        A.ts BETWEEN C.ts - INTERVAL '10' MINUTE AND C.ts + INTERVAL '10' MINUTE