我正在尝试遵循文档并创建一个表函数来“展平”一些数据。使用 进行展平时,表函数似乎工作正常joinLateral
。但是在使用时leftOuterJoinLateral
,我收到以下错误。我正在使用 Scala 并尝试了 Table API 和 SQL,结果相同:
原因:java.lang.NullPointerException:空结果不能存储在案例类中。
这是我的工作:
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.functions.TableFunction
object example_job{
// Split the List[Int] into multiple rows
class Split() extends TableFunction[Int] {
def eval(nums: List[Int]): Unit = {
nums.foreach(x =>
if(x != 3) {
collect(x)
})
}
}
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.createLocalEnvironment()
val tableEnv = StreamTableEnvironment.create(env)
val splitMe = new Split()
// Create some dummy data
val events: DataStream[(String, List[Int])] = env.fromElements(("simon", List(1,2,3)), ("jessica", List(3)))
val table = tableEnv.fromDataStream(events, 'name, 'numbers)
.leftOuterJoinLateral(splitMe('numbers) as 'number)
.select('name, 'number)
table.toAppendStream[(String, Int)].print()
env.execute("Flink jira ticket example")
}
}
当我更改为时.leftOuterJoinLateral
,.joinLateral
我得到了预期的结果:
(simon,1)
(simon,2)
使用时,.leftOuterJoinLateral
我会期望类似:
(simon,1)
(simon,2)
(simon,null) // or (simon, None)
(jessica,null) // or (jessica, None)
似乎这可能是 Scala API 的错误?我想在提出罚单之前先检查这里,以防我做一些愚蠢的事情!