flink.api.table.TableException: 类型不受支持

Question

我正在尝试在 Scala 中使用 Flink Table API。编译时没有错误，但是我运行我的 flink 集群中的作业：flink.api.table.TableException: Type is not supported:<GenericType<java.lang.Object>

我的 Maven 依赖项：

<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

更新：我认为这可能与 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.

感谢任何帮助。谢谢。

Answer 1

Flink 的 Table API 不支持 1.1-SNAPSHOT 中包含 GenericType 的字段。有一个 Pull Request 实现了这个特性。很有可能会包含在Flink 1.1版本中。

关于你的第二个例外：这个例外基本上是自我解释的。到目前为止，您不能对流进行聚合。但是，StreamSQL 在 development.

下

flink.api.table.TableException: 类型不受支持

flink.api.table.TableException: Type is not supported

scala

apache-flink

flink-streaming