Spark Streaming: среднее за все время

Question

Я написал приложение Spark Streaming, которое получает значения температуры и вычисляет среднюю температуру за все время. Для этого я использовал JavaPairDStream.updateStateByKey транзакция для расчета на устройство (разделена ключом пары). Для отслеживания состояния я использую StatCounter класс, который хранит все значения температуры как двойные и пересчитывает среднее значение каждого потока, вызывая StatCounter.mean метод. Вот моя программа:

РЕДАКТИРОВАННЫЙ МОЙ КОД: СЕЙЧАС ИСПОЛЬЗУЙТЕ StatCounter

JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));

streamingContext.checkpoint("hdfs://server:8020/spark-history/checkpointing");

JavaReceiverInputDStream<String> ingoingStream = streamingContext.socketTextStream(serverIp, 11833);

JavaDStream<SensorData> sensorDStream = ingoingStream.map(new Function<String, SensorData>() {
    public SensorData call(String json) throws Exception {
        ObjectMapper om = new ObjectMapper();
        return (SensorData)om.readValue(json, SensorData.class);
    }
});

JavaPairDStream<String, Float> temperatureDStream = sensorDStream.mapToPair(new PairFunction<SensorData, String, Float>() {
    public Tuple2<String, Float> call(SensorData sensorData) throws Exception {
        return new Tuple2<String, Float>(sensorData.getIdSensor(), sensorData.getValTemp());
    }
});

JavaPairDStream<String, StatCounter> statCounterDStream = temperatureDStream.updateStateByKey(new Function2<List<Float>, Optional<StatCounter>, Optional<StatCounter>>() {
    public Optional<StatCounter> call(List<Float> newTemperatures, Optional<StatCounter> statsYet) throws Exception {
        StatCounter stats = statsYet.or(new StatCounter());

        for(float temp : newTemperatures) {
            stats.merge(temp);
        }

        return Optional.of(stats);
    }
});

JavaPairDStream<String, Double> avgTemperatureDStream = statCounterDStream.mapToPair(new PairFunction<Tuple2<String,StatCounter>, String, Double>() {
    public Tuple2<String, Double> call(Tuple2<String, StatCounter> statCounterTuple) throws Exception {
        String key = statCounterTuple._1();
        double avgValue = statCounterTuple._2().mean();

        return new Tuple2<String, Double>(key, avgValue);
    }
});

avgTemperatureDStream.print();

Кажется, это работает нормально. Но теперь к вопросу:

Я только что нашел в Интернете пример, который также показывает, как рассчитать среднее значение за все время здесь: https://databricks.gitbooks.io/databricks-spark-reference-applications/content/logs_analyzer/chapter1/total.html

Они используют AtmoicLongs и т. д. для хранения "значений с состоянием" и обновления их в forEachRDD метод.

Теперь у меня вопрос: что является лучшим решением для расчета состояния всего времени в Spark Streaming? Есть ли преимущества / недостатки использования того или иного способа? Спасибо!