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ストリーミングと同じです。

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 

結果でこのジョブの大きな入力データでパフォーマンスが低下しています。

これをどのように修正して最適な実行計画を得ることができますか？

Flest Table APIとSQLは実験的な機能で、 は次のバージョンで修正される可能性があります。あなたはTableがDataSetまたはDataStreamに変換またはTableSinkに書き込むたびに現在の状態では

Answer 1

、表のAPIは、全体のクエリを変換します。プログラムでは、writeToSinkの3回を呼び出します。これは、完全なクエリが翻訳されるたびに呼び出されることを意味します。

しかし、完全なクエリは何ですか？ Tableには、すべてのTable API演算子が適用されています。 TableをTableEnvironmentに登録すると、基本的にビューとして登録されます。つまり、その定義（テーブルを定義するすべての演算子）のみが登録されます。したがって、writeToSinkを2回目と3回目に呼び出すと、これらの演算子が再度変換されます。

あなたはDataStreamにtableAを翻訳しTableとしてそれを登録する代わりにTableEnvironmentでDataStreamを登録する場合は、この問題を解決することができます。これは以下のようになります。

私が知っている

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 

、これは表の繰り返し翻訳を回避するための唯一の方法は非常に便利ではありませんが、現時点では。