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)