[dbs-leipzig#1570] put timestamp extraction inside the aggregation gr…

…oup reduce
chrissi16 · Aug 11, 2022 · 58b4817 · 58b4817
1 parent e9fd33e
commit 58b4817
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 38 deletions.
diff --git a/.../src/main/java/org/gradoop/temporal/model/impl/operators/metric/DegreeRangeEvolution.java b/.../src/main/java/org/gradoop/temporal/model/impl/operators/metric/DegreeRangeEvolution.java
@@ -16,22 +16,18 @@
 package org.gradoop.temporal.model.impl.operators.metric;
 
 import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.gradoop.common.model.impl.id.GradoopId;
 import org.gradoop.flink.model.api.operators.UnaryBaseGraphToValueOperator;
 import org.gradoop.flink.model.impl.operators.sampling.functions.VertexDegree;
 import org.gradoop.temporal.model.api.TimeDimension;
 import org.gradoop.temporal.model.impl.TemporalGraph;
-import org.gradoop.temporal.model.impl.operators.metric.functions.ExtractAllTimePointsReduce;
 import org.gradoop.temporal.model.impl.operators.metric.functions.GroupDegreeTreesToAggregateDegrees;
 import org.gradoop.temporal.model.impl.operators.metric.functions.AggregateType;
 import org.gradoop.temporal.model.impl.operators.metric.functions.TransformDeltaToAbsoluteDegreeTree;
 import org.gradoop.temporal.model.impl.operators.metric.functions.BuildTemporalDegreeTree;
 import org.gradoop.temporal.model.impl.operators.metric.functions.FlatMapVertexIdEdgeInterval;
 
 import java.util.Objects;
-import java.util.TreeMap;
 
 /**
  * Operator that calculates the degree range evolution of a temporal graph for the
@@ -61,24 +57,15 @@ public DegreeRangeEvolution(VertexDegree degreeType, TimeDimension dimension) {
 
   @Override
   public DataSet<Tuple2<Long, Integer>> execute(TemporalGraph graph) {
-    DataSet<Tuple2<GradoopId, TreeMap<Long, Integer>>> absoluteDegreeTrees = graph.getEdges()
+    return graph.getEdges()
           // 1) Extract vertex id(s) and corresponding time intervals
           .flatMap(new FlatMapVertexIdEdgeInterval(dimension, degreeType))
           // 2) Group them by the vertex id
           .groupBy(0)
           // 3) For each vertex id, build a degree tree data structure
           .reduceGroup(new BuildTemporalDegreeTree())
           // 4) Transform each tree to aggregated evolution
-          .map(new TransformDeltaToAbsoluteDegreeTree());
-
-    DataSet<Tuple1<Long>> timePoints = absoluteDegreeTrees
-          // 5) extract all timestamps where degree of any vertex changes
-          .reduceGroup(new ExtractAllTimePointsReduce())
-          .distinct();
-
-    return absoluteDegreeTrees
-          // join with interval degree mappings
-          // 6) Merge trees together and calculate aggregation
-          .reduceGroup(new GroupDegreeTreesToAggregateDegrees(AggregateType.RANGE, timePoints));
+          .map(new TransformDeltaToAbsoluteDegreeTree())
+          .reduceGroup(new GroupDegreeTreesToAggregateDegrees(AggregateType.RANGE));
   }
 }
diff --git a/...op/temporal/model/impl/operators/metric/functions/GroupDegreeTreesToAggregateDegrees.java b/...op/temporal/model/impl/operators/metric/functions/GroupDegreeTreesToAggregateDegrees.java
@@ -16,16 +16,13 @@
 package org.gradoop.temporal.model.impl.operators.metric.functions;
 
 import org.apache.flink.api.common.functions.GroupReduceFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.util.Collector;
 import org.gradoop.common.model.impl.id.GradoopId;
 
 import java.util.TreeMap;
 import java.util.SortedSet;
 import java.util.TreeSet;
-import java.util.List;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -41,32 +38,14 @@ public class GroupDegreeTreesToAggregateDegrees
    * The aggregate type to use (min,max,avg).
    */
   private final AggregateType aggregateType;
-  /**
-   * The timestamps where at least one vertex degree changes.
-   */
-  private final SortedSet<Long> timePoints;
 
   /**
    * Creates an instance of this group reduce function.
    *
    * @param aggregateType the aggregate type to use (min,max,avg).
-   * @param timePoints the time points were vertex degrees change.
    */
-  public GroupDegreeTreesToAggregateDegrees(AggregateType aggregateType, DataSet<Tuple1<Long>> timePoints) {
+  public GroupDegreeTreesToAggregateDegrees(AggregateType aggregateType) {
     this.aggregateType = aggregateType;
-
-    List<Tuple1<Long>> tuples;
-    try {
-      tuples = timePoints.collect();
-      this.timePoints = new TreeSet<>();
-
-      for (int i = 0; i < timePoints.count(); i = i + 1) {
-        this.timePoints.add(tuples.get(i).getField(0));
-      }
-    } catch (Exception e) {
-      throw new RuntimeException(e);
-    }
-
   }
 
   @Override
@@ -76,10 +55,12 @@ public void reduce(Iterable<Tuple2<GradoopId, TreeMap<Long, Integer>>> iterable,
     // init necessary maps and set
     HashMap<GradoopId, TreeMap<Long, Integer>> degreeTrees = new HashMap<>();
     HashMap<GradoopId, Integer> vertexDegrees = new HashMap<>();
+    SortedSet<Long> timePoints = new TreeSet<>();
 
     // convert the iterables to a hashmap and remember all possible timestamps
     for (Tuple2<GradoopId, TreeMap<Long, Integer>> tuple : iterable) {
       degreeTrees.put(tuple.f0, tuple.f1);
+      timePoints.addAll(tuple.f1.keySet());
     }
 
     int numberOfVertices = degreeTrees.size();