如何显示数据流中更新的当前累加器值?

kninwzqo  于 2021-06-07  发布在  Kafka
关注(0)|答案(2)|浏览(220)

我正在处理一个javajar。累加器将流值相加。问题是,每次值递增或以特定的周期间隔递增时,我都希望在ui中显示该值。
但是,由于累加器值只能从驱动程序获取,因此在进程完成执行之前,我无法访问该值。你知道如何定期访问这个值吗?
我的代码如下

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传输数据。

pengsaosao

pengsaosao1#

在spark中,只有在调用jssc.star()时,实际代码才会开始执行。现在控件使用spark开始运行循环,所有system.out.println只会被调用一次。不会每次都用循环执行。
对于输出操作,请检查文档
你可以用
print()foreachrdd()另存为对象文本或hadoop文件
希望这有帮助

yrwegjxp

yrwegjxp2#

jssc.start();
while(true) {
    System.out.println("current:" + accum.value());
    Thread.sleep(1000);
}

相关问题