org.apache.flink.table.api.TableException: Arity [3] результата не соответствует числу [2] запрошенного типа - PullRequest
Новая
       6

org.apache.flink.table.api.TableException: Arity [3] результата не соответствует числу [2] запрошенного типа

0 голосов
/ 24 января 2019

У меня есть простой тестовый пример, чтобы узнать, как использовать Table API и case / когда: 

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment

import org.apache.flink.api.scala._
case class Person(name: String, age: Int)
object TableTest {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val te = TableEnvironment.getTableEnvironment(env)
    val ds = env.fromCollection(Seq(Person("a",20), Person("b",40), Person("c", 60)))
    te.registerDataSet("person", ds)

    te.toDataSet[Person](table).print()


    val table = te.sqlQuery(
      """
         select name,age,
             case
                when age <= 20 then 'A'
                when age <=40 then 'B'
                when age <= 60 then 'C'
                else 'D'
            end as age_level
            from person
      """.stripMargin(' '))
    te.toDataSet[Person](table).print()
  }
}

Когда я его запускаю, я сталкиваюсь со следующим исключением: age_level - это вычисляемыйстолбец, я не знаю, почему возникает ошибка 

Exception in thread "main" org.apache.flink.table.api.TableException: Arity [3] of result [ArrayBuffer(String, Integer, String)] does not match the number[2] of requested type [com.flink.table.Person(name: String, age: Integer)].
    at org.apache.flink.table.api.TableEnvironment.generateRowConverterFunction(TableEnvironment.scala:1165)
    at org.apache.flink.table.api.BatchTableEnvironment.getConversionMapper(BatchTableEnvironment.scala:339)
    at org.apache.flink.table.api.BatchTableEnvironment.translate(BatchTableEnvironment.scala:504)
    at org.apache.flink.table.api.BatchTableEnvironment.translate(BatchTableEnvironment.scala:476)
    at org.apache.flink.table.api.scala.BatchTableEnvironment.toDataSet(BatchTableEnvironment.scala:141)
    at com.flink.table.TableTest$.main(TableTest.scala:37)
    at com.flink.table.TableTest.main(TableTest.scala)

1 Ответ

0 голосов
/ 24 января 2019

Проблема в том, что таблица, которую вы хотите преобразовать в DataSet[Person], имеет три атрибута: (name, age, age_level), в то время как класс * Person имеет только два поля: (name, age).

Вы можете реализовать новый класс case 

case class PersonWithAgeLevel(name: String, age: Int, age_level: String)

и преобразовать таблицу в DataSet[PersonWithAgeLevel].

...