我刚开始寻找Spark Streaming的有状态计算解决方案,发现了updateStateByKey()函数。
我想要解决的问题是: 10,000个传感器每分钟产生一个二进制值。
如果一个传感器连续报告的值与上一个不同,我希望能够标记并将其作为状态更改事件发送到Kafka。
我假设在这个例子中可以使用updateStateByKey()函数,但我不完全清楚实现相同功能的推荐方法。
我刚开始寻找Spark Streaming的有状态计算解决方案,发现了updateStateByKey()函数。
我想要解决的问题是: 10,000个传感器每分钟产生一个二进制值。
如果一个传感器连续报告的值与上一个不同,我希望能够标记并将其作为状态更改事件发送到Kafka。
我假设在这个例子中可以使用updateStateByKey()函数,但我不完全清楚实现相同功能的推荐方法。
我假设你将从传感器获取一个(String, Int)的流,其中String是传感器的ID,Int是传感器返回的二进制值。在这个假设下,你可以尝试以下代码:
val sensorData: DStream[(String, Int)] = ???
val state = sensorData.updateStateByKey[(String, Int)](updateFunction _)
def updateFunction(newValues: Seq[(String, Int)], currentValues: Seq[(String, Int)]) = {
val newValuesMap = newValues.toMap
val currentValuesMap = currentValues.toMap
currentValuesMap.keys.foreach ( (id) =>
if(currrentValuesMap.get(id) != newValuesMap.getOrElse(id, -1)) {
//send to Kafka
}
)
Some(newValues)
}
州的生命周期/资源是否可以进行管理?或者它总是在增长?
例如,在此会话示例中,状态将永远增长,对吗?是否有任何方法可以管理它,以便您可以清除/存档仅聚合3个月的数据或其他内容?
.mapWithState()
做同样的过程吗? - vdep