我正在处理一个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传输数据。
2条答案
按热度按时间pengsaosao1#
在spark中,只有在调用jssc.star()时,实际代码才会开始执行。现在控件使用spark开始运行循环,所有system.out.println只会被调用一次。不会每次都用循环执行。
对于输出操作,请检查文档
你可以用
print()foreachrdd()另存为对象文本或hadoop文件
希望这有帮助
yrwegjxp2#