beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amits...@apache.org
Subject [1/2] beam git commit: [BEAM-1250] Remove leaf when materializing PCollection to avoid re-evaluation.
Date Sat, 07 Jan 2017 09:07:29 GMT
Repository: beam
Updated Branches:
  refs/heads/master 418c597c4 -> 70dd54625


[BEAM-1250] Remove leaf when materializing PCollection to avoid re-evaluation.


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

Branch: refs/heads/master
Commit: 8d3d381cfea717939f2f925f07118b27b02ec1c2
Parents: 418c597
Author: Sela <ansela@paypal.com>
Authored: Sat Jan 7 00:03:34 2017 +0200
Committer: Sela <ansela@paypal.com>
Committed: Sat Jan 7 10:41:27 2017 +0200

----------------------------------------------------------------------
 .../beam/runners/spark/translation/EvaluationContext.java     | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/8d3d381c/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java
index a412e31..ec5ad3d 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java
@@ -194,16 +194,15 @@ public class EvaluationContext {
    * @param <T>         Type of elements contained in collection.
    * @return Natively types result associated with collection.
    */
-  public <T> Iterable<T> get(PCollection<T> pcollection) {
-    @SuppressWarnings("unchecked")
-    BoundedDataset<T> boundedDataset = (BoundedDataset<T>) datasets.get(pcollection);
-    Iterable<WindowedValue<T>> windowedValues = boundedDataset.getValues(pcollection);
+  <T> Iterable<T> get(PCollection<T> pcollection) {
+    Iterable<WindowedValue<T>> windowedValues = getWindowedValues(pcollection);
     return Iterables.transform(windowedValues, WindowingHelpers.<T>unwindowValueFunction());
   }
 
   <T> Iterable<WindowedValue<T>> getWindowedValues(PCollection<T>
pcollection) {
     @SuppressWarnings("unchecked")
     BoundedDataset<T> boundedDataset = (BoundedDataset<T>) datasets.get(pcollection);
+    leaves.remove(boundedDataset);
     return boundedDataset.getValues(pcollection);
   }
 


Mime
View raw message