我正在使用Spark Streaming应用中的updateStateByKey函数,以持久化和更新每个key的状态。问题是我想知道在update函数内部的"key"。
input.updateStateByKey(updateStateByKeyOfUsers)
def updateStateByKeyOfUsers(newUsers: Seq[Set[String]],
userStatus: Option[(#####)]
): Option[(#####)] = {
//How to get the "Key"
}
-Tao
input
类型的话会有所帮助吗?根据这个例子,我认为关键在于第55行的t._1
,尽管你的方法签名看起来非常不同。编译通过了吗? - Gábor Bakos