diff options
author | Sean Owen <sowen@cloudera.com> | 2016-06-12 11:44:33 -0700 |
---|---|---|
committer | Reynold Xin <rxin@databricks.com> | 2016-06-12 11:44:33 -0700 |
commit | f51dfe616b24b4234199c98ea857a586a93a889f (patch) | |
tree | 2803e1675f1948670ebc3f042789f4b401aa2b3e /examples/src | |
parent | 50248dcfff3ba79b73323f3a804c1e19a8be6097 (diff) | |
download | spark-f51dfe616b24b4234199c98ea857a586a93a889f.tar.gz spark-f51dfe616b24b4234199c98ea857a586a93a889f.tar.bz2 spark-f51dfe616b24b4234199c98ea857a586a93a889f.zip |
[SPARK-15086][CORE][STREAMING] Deprecate old Java accumulator API
## What changes were proposed in this pull request?
- Deprecate old Java accumulator API; should use Scala now
- Update Java tests and examples
- Don't bother testing old accumulator API in Java 8 (too)
- (fix a misspelling too)
## How was this patch tested?
Jenkins tests
Author: Sean Owen <sowen@cloudera.com>
Closes #13606 from srowen/SPARK-15086.
Diffstat (limited to 'examples/src')
-rw-r--r-- | examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java | 10 |
1 files changed, 5 insertions, 5 deletions
diff --git a/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java b/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java index 0563149448..acbc345243 100644 --- a/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java +++ b/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java @@ -29,7 +29,6 @@ import scala.Tuple2; import com.google.common.io.Files; -import org.apache.spark.Accumulator; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaSparkContext; @@ -41,6 +40,7 @@ import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; +import org.apache.spark.util.LongAccumulator; /** * Use this singleton to get or register a Broadcast variable. @@ -67,13 +67,13 @@ class JavaWordBlacklist { */ class JavaDroppedWordsCounter { - private static volatile Accumulator<Integer> instance = null; + private static volatile LongAccumulator instance = null; - public static Accumulator<Integer> getInstance(JavaSparkContext jsc) { + public static LongAccumulator getInstance(JavaSparkContext jsc) { if (instance == null) { synchronized (JavaDroppedWordsCounter.class) { if (instance == null) { - instance = jsc.accumulator(0, "WordsInBlacklistCounter"); + instance = jsc.sc().longAccumulator("WordsInBlacklistCounter"); } } } @@ -158,7 +158,7 @@ public final class JavaRecoverableNetworkWordCount { final Broadcast<List<String>> blacklist = JavaWordBlacklist.getInstance(new JavaSparkContext(rdd.context())); // Get or register the droppedWordsCounter Accumulator - final Accumulator<Integer> droppedWordsCounter = + final LongAccumulator droppedWordsCounter = JavaDroppedWordsCounter.getInstance(new JavaSparkContext(rdd.context())); // Use blacklist to drop words and use droppedWordsCounter to count them String counts = rdd.filter(new Function<Tuple2<String, Integer>, Boolean>() { |