How to create NULLable Flink table columns from Scala case classes that contain Option types

Viewed 451

I would like to create a DataSet (or DataStream) from a collection of case classes that contain Option values. In the created table columns resulting from Option values should either contain NULL or the actual primitive value.

This is what I tried:

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

object OptionExample {
  case class Event(time: Timestamp, id: String, value: Option[Int])

  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val tEnv = TableEnvironment.getTableEnvironment(env)

    val data = env.fromCollection(Seq(
      Event(Timestamp.valueOf("2018-01-01 00:01:00"), "a", Some(3)),
      Event(Timestamp.valueOf("2018-01-01 00:03:00"), "a", None),
      Event(Timestamp.valueOf("2018-01-01 00:03:00"), "b", Some(7)),
      Event(Timestamp.valueOf("2018-01-01 00:02:00"), "a", Some(5))
    ))

    val table = tEnv.fromDataSet(data)
    table.printSchema()
//    root
//    |-- time: Timestamp
//    |-- id: String
//    |-- value: Option[Integer]

    val result = table
      .groupBy('id)
      .select('id, 'value.avg as 'averageValue)

    // Print results
    val ds: DataSet[Row] = result.toDataSet
    ds.print()
  }
}

But this causes an Exception in the aggregation part...

org.apache.flink.table.api.ValidationException: Expression avg('value) failed on input check: avg requires numeric types, get Option[Integer] here

...so with this approach Option does not get converted into a numeric type with NULLs as described above.

How can I achieve this with Flink?

(I'm coming from Apache Spark, there Datasets created from case classes with Options have this behaviour. I would like to achieve something similar with Flink)

0 Answers
Related