From c5075aa5809358c16b6b6aafaf0294d1fb87cbb9 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 8 Apr 2014 05:09:24 -0700 Subject: [PATCH] If guava 14 had iterables --- .../apache/spark/rdd/PairRDDFunctions.scala | 9 +++---- .../java/org/apache/spark/JavaAPISuite.java | 26 +++---------------- .../scala/org/apache/spark/FailureSuite.scala | 3 +-- .../apache/spark/examples/JavaPageRank.java | 10 +++---- .../java/org/apache/spark/Java8APISuite.java | 19 ++++---------- 5 files changed, 16 insertions(+), 51 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala index f2127b5384e94..a92a84b5342d1 100644 --- a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala +++ b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala @@ -298,8 +298,7 @@ class PairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) */ def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = { this.cogroup(other, partitioner).flatMapValues { case (vs, ws) => - val wlist = ws.toList - for (v <- vs; w <- wlist.iterator) yield (v, w) + for (v <- vs; w <- ws) yield (v, w) } } @@ -314,8 +313,7 @@ class PairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) if (ws.isEmpty) { vs.map(v => (v, None)) } else { - val wlist = ws.toList - for (v <- vs; w <- wlist.iterator) yield (v, Some(w)) + for (v <- vs; w <- ws) yield (v, Some(w)) } } } @@ -332,8 +330,7 @@ class PairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) if (vs.isEmpty) { ws.map(w => (None, w)) } else { - val wlist = ws.toList - for (v <- vs; w <- wlist) yield (Some(v), w) + for (v <- vs; w <- ws) yield (Some(v), w) } } } diff --git a/core/src/test/java/org/apache/spark/JavaAPISuite.java b/core/src/test/java/org/apache/spark/JavaAPISuite.java index 81c4a0223a465..e82caaac8a042 100644 --- a/core/src/test/java/org/apache/spark/JavaAPISuite.java +++ b/core/src/test/java/org/apache/spark/JavaAPISuite.java @@ -24,6 +24,7 @@ import scala.Tuple2; import com.google.common.collect.Lists; +import com.google.collect.Iterables; import com.google.common.base.Optional; import com.google.common.base.Charsets; import com.google.common.io.Files; @@ -78,35 +79,14 @@ public int compare(Integer a, Integer b) { } } - private int iteratorSize(Iterator a) { - int size = 0; - while (a.hasNext()) { - size++; - a.next(); - } - return size; - } private int iterableSize(Iterable a) { - return iteratorSize(a.iterator()); + return Iterables.size(a.iterator()); } - private String iteratorStr(Iterator a) { - StringBuilder str = new StringBuilder(); - str.append("["); - while (a.hasNext()) { - str.append(a.next().toString()); - if (a.hasNext()) { - str.append(", "); - } - } - str.append("]"); - return str.toString(); - } - private String iterableStr(Iterable a) { - return iteratorStr(a.iterator()); + return Iterables.toString(a.iterator()); } diff --git a/core/src/test/scala/org/apache/spark/FailureSuite.scala b/core/src/test/scala/org/apache/spark/FailureSuite.scala index d6ee130f9fd13..12dbebcb28644 100644 --- a/core/src/test/scala/org/apache/spark/FailureSuite.scala +++ b/core/src/test/scala/org/apache/spark/FailureSuite.scala @@ -72,8 +72,7 @@ class FailureSuite extends FunSuite with LocalSparkContext { throw new Exception("Intentional task failure") } } - val vHead = v.iterator.next() - (k, vHead * vHead) + (k, v.head * v.head) }.collect() FailureSuiteState.synchronized { assert(FailureSuiteState.tasksRun === 4) diff --git a/examples/src/main/java/org/apache/spark/examples/JavaPageRank.java b/examples/src/main/java/org/apache/spark/examples/JavaPageRank.java index cffc747b3c286..c0525f887e7cd 100644 --- a/examples/src/main/java/org/apache/spark/examples/JavaPageRank.java +++ b/examples/src/main/java/org/apache/spark/examples/JavaPageRank.java @@ -17,7 +17,10 @@ package org.apache.spark.examples; + import scala.Tuple2; + +import com.google.collect.Iterables; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; @@ -90,12 +93,7 @@ public Double call(Iterable rs) { .flatMapToPair(new PairFlatMapFunction, Double>, String, Double>() { @Override public Iterable> call(Tuple2, Double> s) { - int urlCount = 0; - Iterator urls = s._1.iterator(); - while (urls.hasNext()) { - urls.next(); - urlCount++; - } + int urlCount = Iterables.size(s._1); List> results = new ArrayList>(); for (String n : s._1) { results.add(new Tuple2(n, s._2() / urlCount)); diff --git a/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java b/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java index 0e409f30a2881..9c45863f8bae9 100644 --- a/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java +++ b/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java @@ -23,6 +23,7 @@ import scala.Tuple2; +import com.google.collections.Iterables; import com.google.common.base.Optional; import com.google.common.io.Files; import org.apache.hadoop.io.IntWritable; @@ -60,16 +61,6 @@ public void tearDown() { System.clearProperty("spark.driver.port"); } - private int iterableSize(Iterable a) { - int count = 0; - Iterator aItr = a.iterator(); - while (aItr.hasNext()) { - aItr.next(); - count++; - } - return count; - } - @Test public void foreachWithAnonymousClass() { foreachCalls = 0; @@ -97,13 +88,13 @@ public void groupBy() { Function isOdd = x -> x % 2 == 0; JavaPairRDD> oddsAndEvens = rdd.groupBy(isOdd); Assert.assertEquals(2, oddsAndEvens.count()); - Assert.assertEquals(2, iterableSize(oddsAndEvens.lookup(true).get(0))); // Evens - Assert.assertEquals(5, iterableSize(oddsAndEvens.lookup(false).get(0))); // Odds + Assert.assertEquals(2, Iterables.size(oddsAndEvens.lookup(true).get(0))); // Evens + Assert.assertEquals(5, Iterables.size(oddsAndEvens.lookup(false).get(0))); // Odds oddsAndEvens = rdd.groupBy(isOdd, 1); Assert.assertEquals(2, oddsAndEvens.count()); - Assert.assertEquals(2, iterableSize(oddsAndEvens.lookup(true).get(0))); // Evens - Assert.assertEquals(5, iterableSize(oddsAndEvens.lookup(false).get(0))); // Odds + Assert.assertEquals(2, Iterables.size(oddsAndEvens.lookup(true).get(0))); // Evens + Assert.assertEquals(5, Iterables.size(oddsAndEvens.lookup(false).get(0))); // Odds } @Test