org.apache.flink.api.table.TableException： "フィールド参照式のエイリアスが必要です"

Question

ストリームのSQLクエリを評価するために、1.1-SNAPSHOTバージョンのApache FlinkのテーブルAPIを使用しています。続き

は私のコードです：私はこのプログラムを実行すると

private static final int MAX_RACK_ID = 10; 
private static final long PAUSE = 100; 
private static final double TEMP_STD = 20; 
private static final double TEMP_MEAN = 80; 

public static void main(String[] args) 
{ 
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
    StreamTableEnvironment tableEnv=TableEnvironment.getTableEnvironment(env); 

    DataStream<MonitoringEvent> dstream = env.addSource(new MonitoringEventSource(MAX_RACK_ID, PAUSE, TEMP_STD, TEMP_MEAN)); 
    tableEnv.registerDataStream("TemperatureData", dstream,"rackid,temperature,timestamp"); 

    Table tab1 = tableEnv.sql("select STREAM rackid,temperature,timestamp from TemperatureData where temperature>=100"); 
    DataStream<TemperatureEvent>tempstream=tableEnv.toDataStream(tab1, TemperatureEvent.class); 
    tempstream.print(); 
} 

、それは次の例外がスローされます。

Exception in thread "main" org.apache.flink.api.table.TableException: Alias on field reference expression expected. 
    at org.apache.flink.api.table.TableEnvironment$$anonfun$4.apply(TableEnvironment.scala:299) 
    at org.apache.flink.api.table.TableEnvironment$$anonfun$4.apply(TableEnvironment.scala:292) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) 
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:244) 
    at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108) 
    at org.apache.flink.api.table.TableEnvironment.getFieldInfo(TableEnvironment.scala:292) 
    at org.apache.flink.api.table.StreamTableEnvironment.registerDataStreamInternal(StreamTableEnvironment.scala:212) 
    at org.apache.flink.api.java.table.StreamTableEnvironment.registerDataStream(StreamTableEnvironment.scala:130) 
    at com.yash.flink.Program.main(Program.java:31) 

私はいくつか質問があります：何ですか

• をApache FlinkのTable APIを使ってストリーム上にSQLクエリを書く方法？
• Flinkでこのクエリを実装するにはどうすればよいですか？
• これはFlink's Table APIのバグですか？

Answer 1

テーブルAPIには制限事項があります。この問題は、DataStream<MonitoringEvent>をテーブルとして登録することによって発生します。あなたはする必要があります

tableEnv.registerDataStream(
    "TemperatureData", 
    dstream, 
    "rackid AS rackid, temperature AS temperature, timestamp AS timestamp" 
); 

それを動作させる必要があります。 Stream SQLがFlink 1.1.0でリリースされる前に問題が解決されていることを確認します。