Flink：执行计划中的表 API 复制运算符

Question

我使用 Flink 1.2.0 Table API 来处理一些流数据。以下是我的代码：

val dataTable = myDataStream
// table A
val tableA = dataTable
  .window(Tumble over 5.minutes on 'rowtime as 'w)
  .groupBy("w, group1, group2")
  .select("w.start as time, group1, group2, data1.sum as data1, data2.sum as data2")
tableEnv.registerTable("tableA", tableA)
// table A sink
tableA.writeToSink(sinkTableA)
//...
// I shoul get some other outputs from TableA output
//...
val dataTable = tableEnv.ingest("tableA")
// table result1
val result1 = dataTable
 .window(Tumble over 5.minutes on 'rowtime as 'w)
 .groupBy("w, group1")
 .select("w.start as time, group1, data1.sum as data1")
// result1 sink
result2.writeToSink(sinkResult1)
// table result2
val result2 = dataTable
 .window(Tumble over 5.minutes on 'rowtime as 'w)
 .groupBy("w, group2")
 .select("w.start as time, group2, data2.sum as data1")
// result2 sink
result2.writeToSink(sinkResult2)

我等待在 flink 执行计划中获取这棵树。 与我在其他 Flink 工作中的 Flink Streaming 相同。

DataStream_Operators -> TableA_Operators -> TableA_Sink
                                        |-> Result1_Operators -> Result1_Sink
                                        |-> Result2_Operators -> Result2_Sink

但是，我用 TableA 的 3 个相同操作符得到了这个！

DataStream_Operators -> TableA_Operators -> TableA_Sink
                   |-> Copy_of_TableA_Operators -> Result1_Operators -> Result1_Sink
                   |-> Copy_of_TableA_Operators -> Result2_Operators -> Result2_Sink

我在这项工作的大量输入数据中表现不佳。

我该如何解决这个问题并获得最佳执行计划？

我不明白，Flink Table API 和 SQL 是什么实验性功能和 也许它会在下一个版本中修复。

Answer 1

在当前状态下，每当您将Table 转换为DataSet 或DataStream 或将其写入TableSink 时，Table API 都会转换整个查询。在您的程序中，您调用了三次writeToSink，这意味着每次翻译完整的查询。

但是完整的查询是什么？所有表 API 运算符都已应用于 Table。当您在TableEnvironment 中注册Table 时，它基本上注册为视图，即仅注册其定义（定义表的所有运算符）。因此，当您第二次和第三次调用writeToSink 时，这些运算符会被再次翻译。

如果您将tableA 转换为DataStream 并在TableEnvironment 中注册DataStream，而不是将其注册为Table，则可以解决此问题。如下所示：

val tableA = ...
val streamA = tableA.toDataStream[X] // X should be a case class for rows of tableA
val tableEnv.registerDataStream("tableA", streamA)

tableEnv.ingest("tableA").writeToSink(sinkTableA) // emit tableA by ingesting the registered DataStream

我知道，这不是很方便，但目前是避免重复翻译表格的唯一方法。