beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From stasle...@apache.org
Subject [1/2] beam git commit: [BEAM-1304] Checking for nullity before trying to obtain an aggregator's value.
Date Sun, 05 Feb 2017 14:02:46 GMT
Repository: beam
Updated Branches:
  refs/heads/master e5afbb27f -> c442ef81a


[BEAM-1304] Checking for nullity before trying to obtain an aggregator's value.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/882c654b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/882c654b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/882c654b

Branch: refs/heads/master
Commit: 882c654b1a8aefd2e4281d786448734731db7816
Parents: e5afbb2
Author: Stas Levin <staslevin@gmail.com>
Authored: Sun Feb 5 12:17:35 2017 +0200
Committer: Stas Levin <staslevin@gmail.com>
Committed: Sun Feb 5 15:51:18 2017 +0200

----------------------------------------------------------------------
 .../runners/spark/aggregators/NamedAggregators.java    |  6 ++++--
 .../aggregators/metrics/sink/NamedAggregatorsTest.java | 13 +++++++++++++
 2 files changed, 17 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/882c654b/runners/spark/src/main/java/org/apache/beam/runners/spark/aggregators/NamedAggregators.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/aggregators/NamedAggregators.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/aggregators/NamedAggregators.java
index b5aec32..c876c07 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/aggregators/NamedAggregators.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/aggregators/NamedAggregators.java
@@ -68,10 +68,12 @@ public class NamedAggregators implements Serializable {
    * @param name      Name of aggregator to retrieve.
    * @param typeClass Type class to cast the value to.
    * @param <T>       Type to be returned.
-   * @return the value of the aggregator associated with the specified name
+   * @return the value of the aggregator associated with the specified name,
+   * or <code>null</code> if the specified aggregator could not be found.
    */
   public <T> T getValue(String name, Class<T> typeClass) {
-    return typeClass.cast(mNamedAggregators.get(name).render());
+    final State<?, ?, ?> state = mNamedAggregators.get(name);
+    return state != null ? typeClass.cast(state.render()) : null;
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/beam/blob/882c654b/runners/spark/src/test/java/org/apache/beam/runners/spark/aggregators/metrics/sink/NamedAggregatorsTest.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/aggregators/metrics/sink/NamedAggregatorsTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/aggregators/metrics/sink/NamedAggregatorsTest.java
index 3b5dd21..8646510 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/aggregators/metrics/sink/NamedAggregatorsTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/aggregators/metrics/sink/NamedAggregatorsTest.java
@@ -28,10 +28,13 @@ import java.util.List;
 import java.util.Set;
 import org.apache.beam.runners.spark.SparkPipelineOptions;
 import org.apache.beam.runners.spark.aggregators.ClearAggregatorsRule;
+import org.apache.beam.runners.spark.aggregators.SparkAggregators;
 import org.apache.beam.runners.spark.examples.WordCount;
+import org.apache.beam.runners.spark.translation.SparkContextFactory;
 import org.apache.beam.runners.spark.translation.streaming.utils.SparkTestPipelineOptions;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.MapElements;
@@ -95,4 +98,14 @@ public class NamedAggregatorsTest {
     assertThat(InMemoryMetrics.<Double>valueOf("emptyLines"), is(1d));
 
   }
+
+  @Test
+  public void testNonExistingAggregatorName() throws Exception {
+    final SparkPipelineOptions options = PipelineOptionsFactory.as(SparkPipelineOptions.class);
+    final Long valueOf =
+        SparkAggregators.valueOf(
+            "myMissingAggregator", Long.class, SparkContextFactory.getSparkContext(options));
+
+    assertThat(valueOf, is(nullValue()));
+  }
 }


Mime
View raw message