I've just started looking around for a solution for stateful computation with Spark Streaming when I came across the updateStateByKey() function. The problem I'm trying to solve: 10,000 sensors produce a binary value every minute. If consecutive values a sensor reports are different from each other, I would like to flag that and send it down Kaf…

gistfile1.txt

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)
}