在 Spark Streaming 中使用 updateStateByKey() 从原始事件流生成状态更改流

Question

当我遇到 updateStateByKey() 函数时，我刚刚开始寻找 Spark Streaming 有状态计算的解决方案。

我试图解决的问题是：10,000 个传感器每分钟产生一个二进制值。

如果传感器报告的连续值彼此不同，我想标记它并将其作为状态更改事件发送到 Kafka。

我的假设是 updateStateByKey() 可以用于此示例，但是我并不完全了解实现相同方法的推荐方法。

Answer 1

我假设您将从传感器获得 (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)
}