Как рассчитать агрегации в окне, когда показания датчика не отправляются, если они не изменились с момента последнего события?

Question

Как я могу вычислить агрегации в окне от датчика, когда новые события отправляются, только если значение датчика изменилось с момента последнего события?Показания датчика снимаются в фиксированное время, например, каждые 5 секунд, но они передаются только в том случае, если показания изменяются с момента последнего чтения.

Итак, если я хочу создать среднее значение signal_stength для каждого устройства:

eventsDF = ... 
avgSignalDF = eventsDF.groupBy("deviceId").avg("signal_strength")

Например, события, отправленные устройством за одноминутное окно:

event_time  device_id  signal_strength
12:00:00    1          5
12:00:05    1          4
12:00:30    1          5
12:00:45    1          6
12:00:55    1          5

Тот же набор данных с событиями, которые на самом деле не отправляются, заполнены:

event_time  device_id  signal_strength
12:00:00    1          5
12:00:05    1          4
12:00:10    1          4
12:00:15    1          4
12:00:20    1          4
12:00:25    1          4
12:00:30    1          5
12:00:35    1          5
12:00:40    1          5
12:00:45    1          6
12:00:50    1          6
12:00:55    1          5

signal_strength sum равно 57, а avg равно 57/12

Каким образом эти недостающие данные могут быть выведены с помощью искровой структурированной потоковой передачи, а среднее значение рассчитано из выведенных значений?

Примечание : Я использовал среднее значение в качестве примера агрегации, но решение должно работать для любой функции агрегации.

Answer 1

Математически эквивалентно средневзвешенной задаче на основе длительности:

avg=(signal_strength*duration)/60

Задача здесь состоит в том, чтобы получить длительность для каждого сигнала, один вариант здесь для каждой микропакета, собрать результат в драйвере, затем онЭто все статистические проблемы, чтобы получить длительность, вы можете сделать сдвиг влево по времени начала, затем вычесть, что-то вроде этого:

window.start.leftShift(1)-window.start

, что даст вам:

event_time  device_id  signal_strength duration
12:00:00    1          5                  5(5-0)
12:00:05    1          4                  25(30-5)
12:00:30    1          5                  15(45-30)
12:00:45    1          6                  10(55-45)
12:00:55    1          5                  5 (60-55)

(5*5+4*25+5*15+6*10+5*5)/60=57/12

По состоянию наSpark структурированная потоковая передача 2.3.2, вам нужно написать собственный настраиваемый приемник, чтобы собирать результаты каждого этапа в драйвер и выполнять математическую работу таким образом.

Answer 2

РЕДАКТИРОВАНИЕ:

Я изменил логику, чтобы вычислять среднее значение только из отфильтрованного dataframe, чтобы оно учитывало пропуски.

//input structure
case class StreamInput(event_time: Long, device_id: Int, signal_strength: Int)
//columns for which we want to maintain state
case class StreamState(prevSum: Int, prevRowCount: Int, prevTime: Long, prevSignalStrength: Int, currentTime: Long, totalRow: Int, totalSum: Int, avg: Double)
//final result structure
case class StreamResult(event_time: Long, device_id: Int, signal_strength: Int, avg: Double)

val filteredDF = ???  //get input(filtered rows only)

val interval = 5  // event_time interval

// using .mapGroupsWithState to maintain state for runningSum & total row count till now

// you need to set the timeout threshold to indicate how long you wish to maintain the state
val avgDF = filteredDF.groupByKey(_.device_id)
  .mapGroupsWithState[StreamState, StreamResult](GroupStateTimeout.NoTimeout()) {

  case (id: Int, eventIter: Iterator[StreamInput], state: GroupState[StreamState]) => {
    val events = eventIter.toSeq

    val updatedSession = if (state.exists) {
      //if state exists update the state with the new values
      val existingState = state.get

      val prevTime = existingState.currentTime
      val currentTime = events.map(x => x.event_time).last
      val currentRowCount = (currentTime - prevTime)/interval
      val rowCount = existingState.rowCount + currentRowCount.toInt
      val currentSignalStength = events.map(x => x.signal_strength).last

      val total_signal_strength = currentSignalStength + 
        (existingState.prevSignalStrength * (currentRowCount -1)) + 
        existingState.total_signal_strength

      StreamState(
        existingState.total_signal_strength,
        existingState.rowCount,
        prevTime,
        currentSignalStength,
        currentTime,
        rowCount,
        total_signal_strength.toInt,
        total_signal_strength/rowCount.toDouble
      )

    } else {
      // if there are no earlier state
      val runningSum = events.map(x => x.signal_strength).sum
      val size = events.size.toDouble
      val currentTime = events.map(x => x.event_time).last
      StreamState(0, 1, 0, runningSum, currentTime, 1, runningSum, runningSum/size)
    }

    //save the updated state
    state.update(updatedSession)
    StreamResult(
      events.map(x => x.event_time).last,
      id,
      events.map(x => x.signal_strength).last,
      updatedSession.avg
    )
  }
}

val result = avgDF
  .writeStream
  .outputMode(OutputMode.Update())
  .format("console")
  .start

Идея состоит в том, чтобы вычислить два новых столбца:

1. totalRowCount: промежуточный итог количества строк, которые должны присутствовать, если вы не отфильтровали.
2. total_signal_strength: текущий итог signal_strength до сих пор.(это также включает пропущенные итоги строк).

Его рассчитывается по:

total_signal_strength = 
  current row's signal_strength  +  
  (total_signal_strength of previous row * (rowCount -1)) + 
  //rowCount is the count of missed rows computed by comparing previous and current event_time.
  previous total_signal_strength

формат промежуточного состояния:

+----------+---------+---------------+---------------------+--------+
|event_time|device_id|signal_strength|total_signal_strength|rowCount|
+----------+---------+---------------+---------------------+--------+
|         0|        1|              5|                    5|       1|
|         5|        1|              4|                    9|       2|
|        30|        1|              5|                   30|       7|
|        45|        1|              6|                   46|      10|
|        55|        1|              5|                   57|      12|
+----------+---------+---------------+---------------------+--------+

окончательный вывод:

+----------+---------+---------------+-----------------+
|event_time|device_id|signal_strength|              avg|
+----------+---------+---------------+-----------------+
|         0|        1|              5|              5.0|
|         5|        1|              4|              4.5|
|        30|        1|              5|4.285714285714286|
|        45|        1|              6|              4.6|
|        55|        1|              5|             4.75|
+----------+---------+---------------+-----------------+