Как работает Spark Streaming countByValueAndWindow?

Question

У меня есть приложение Spark Streaming, которое обрабатывает поток событий щелчка веб-сайта. Каждое событие имеет свойство, содержащее GUID, который идентифицирует сеанс пользователя, к которому принадлежит событие.

Мое приложение подсчитывая число событий, произошедших на каждой сессии, используя оконное:

def countEvents(kafkaStream: DStream[(String, Event)]): DStream[(String, Session)] = { 

    // Get a list of the session GUIDs from the events 
    val sessionGuids = kafkaStream 
    .map(_._2) 
    .map(_.getSessionGuid) 

    // Count up the GUIDs over our sliding window 
    val sessionGuidCountsInWindow = sessionGuids.countByValueAndWindow(Seconds(60), Seconds(1)) 

    // Create new session objects with the event count 
    sessionGuidCountsInWindow 
    .map({ 
     case (guidS, eventCount) => 
     guidS -> new Session().setGuid(guidS).setEventCount(eventCount) 
    }) 
} 

Насколько я понимаю, что функция countByValueAndWindow только подсчет значений в DStream на которых называется функция , Другими словами, в коде выше вызов countByValueAndWindow должен возвращать количество событий только для GUID сеанса в sessionGuids DStream, на котором мы вызываем эту функцию.

Но я наблюдаю что-то другое; вызов countByValueAndWindow возвращает count для GUID сеанса, которые не входят в sessionGUID. Кажется, что он возвращает counts для GUID сеанса, которые были обработаны в предыдущих партиях. Я просто не понимаю, как работает эта функция? Я не смог найти что-либо на пути полезной документации в Интернете.

Answer 1

Мой коллега, который намного более разбирается в способах искры, чем я, помог мне в этом. По-видимому, я неправильно понял, как работает функция countByValueAndWindow. Я думал, что он вернет только значения для значений в DStream, для которых вы вызываете функцию. Но на самом деле он возвращает значения для всех значений во всем окне. Чтобы решить мою проблему, я просто выполняю соединение между моим входом DStream и DStream в результате операции countByValueAndWindow. Таким образом, я получаю результаты только для значений в моем входе DStream.