我正在尝试在Scala中使用Flink Table API。编译时间没有错误,但是qhen我正在flink群集中运行工作: flink.api.table.TableException: Type is not supported:<GenericType<java.lang.Object>
我的小依赖性:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table_2.11</artifactId>
<version>1.1-SNAPSHOT</version>
</dependency>
我的进口:
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.table.{Row, Table, TableEnvironment}
我的代码:
// odo[(Long,String,Double,Long)]
val inputTable = odo.toTable(tableEnv,'ts,'ty, 'vl, 'dv)
val resultStream: Table = inputTable.where('ty === "Odometer").select('dv)
resultStream.toDataStream[Row].print
update :我认为这可能是关于Flink版本(1.0.3)的,因为当我这样做时:
val inputTable = odo.toTable(tableEnv, 'ts, 'ty, 'vl, 'dv)
val result = inputTable.select('dv,'vl.sum).where('dv == 111)
result.toDataStream[Row].print()
我还有另一个例外: org.apache.flink.api.table.TableException: Aggregate on stream tables is currently not supported.
任何帮助将不胜感激。谢谢。
flink的表API不支持1.1-SNAPSHOT
中包含GenericType
的字段。有一个实现此功能的拉动请求。它很可能包含在Flink 1.1释放中。
关于您的第二个例外:例外基本上是自我解释。到目前为止,您无法在流上进行聚合。但是,StreamSQL正在开发中。