flink.api.table.TableException: Тип не поддерживается

Question

Я пытаюсь использовать API Flink Table в scala. Нет ошибок при составлении времени, но qhen i'm выполнения задания в моем 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

API таблицы Flink не поддерживает поля, содержащие GenericType в 1.1-SNAPSHOT. Существует Pull Request, который реализует эту функцию. Весьма вероятно, что он будет содержаться в выпуске Flink 1.1.

Что касается вашего второго исключения: исключение в основном объясняет сам. Пока вы не можете делать скопления на потоках. Однако StreamSQL находится под development.