0

我编写了一个 Spark Streaming 应用程序，它接收温度值并计算所有时间的平均温度。为此，我使用JavaPairDStream.updateStateByKey事务来计算每个设备（由 Pair 的密钥分隔）。对于状态跟踪，我使用该类，它将所有温度值保存为双精度值，并通过调用该方法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

他们使用AtmoicLongsetc. 来存储“有状态值”并在forEachRDD方法中更新它们。

我现在的问题是：对于 Spark Streaming 中所有时间的有状态计算，更好的解决方案是什么？使用一种或另一种方式有什么优点/缺点吗？谢谢！

4

0 回答 0