java - 如何显示在 DStream 中更新的当前累加器值？

Question

我正在处理一个 java jar。累加器将流值相加。问题是，我想在每次递增时或在特定的周期间隔内在我的 UI 中显示该值。

但是，由于累加器的值只能从 Driver 程序中获取，因此在进程完成执行之前我无法访问该值。关于如何定期访问此值的任何想法？

我的代码如下

package com.spark;

import java.util.HashMap;
import java.util.Map;

import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import scala.Tuple2;

public class KafkaSpark {

    /**
     * @param args
     */
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("Simple Application");
        conf.setMaster("local");
        JavaStreamingContext jssc = new JavaStreamingContext(conf,
                new Duration(5000));
        final Accumulator<Integer> accum = jssc.sparkContext().accumulator(0);
        Map<String, Integer> topicMap = new HashMap<String, Integer>();
        topicMap.put("test", 1);
        JavaPairDStream<String, String> lines = KafkaUtils.createStream(jssc,
                "localhost:2181", "group1", topicMap);

        JavaDStream<Integer> map = lines
                .map(new Function<Tuple2<String, String>, Integer>() {

                    public Integer call(Tuple2<String, String> v1)
                            throws Exception {
                        if (v1._2.contains("the")) {
                            accum.add(1);
                            return 1;
                        }
                        return 0;
                    }
                });

        map.print();
        jssc.start();
        jssc.awaitTermination();
        System.out.println("*************" + accum.value());
        System.out.println("done");
    }
}

我正在使用 Kafka 流式传输数据。

Answer 1

在 spark 中，只有当 jssc.star() 被调用时，实际代码才会开始执行。现在控件与 spark 一起开始运行循环，所有你的 system.out.println 将只被调用一次。并且不会每次都循环执行。

对于输出操作，请检查 documentation

你可以使用

打印()forEachRDD()另存为对象文本或hadoop文件

希望对你有帮助