#1570 degree range evolution

gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/DegreeRangeEvolution.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright © 2014 - 2021 Leipzig University (Database Research Group)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.gradoop.temporal.model.impl.operators.metric;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.tuple.Tuple3;
+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.BuildTemporalDegreeTree;
+import org.gradoop.temporal.model.impl.operators.metric.functions.FlatMapVertexIdEdgeInterval;
+import org.gradoop.temporal.model.impl.operators.metric.functions.GroupDegreeTreesToDegreeRange;
+import org.gradoop.temporal.model.impl.operators.metric.functions.TransformDeltaToAbsoluteDegreeTree;
+import org.gradoop.temporal.model.impl.operators.metric.functions.MapDegreesToInterval;
+
+
+import java.util.Objects;
+
+/**
+ * Operator that calculates the degree range evolution of a temporal graph for the
+ * whole lifetime of the graph.
+ */
+public class DegreeRangeEvolution implements UnaryBaseGraphToValueOperator<TemporalGraph, DataSet<Tuple3<Long, Long, Float>>> {
+  /**
+   * The time dimension that will be considered.
+   */
+  private final TimeDimension dimension;
+
+  /**
+   * The degree type (IN, OUT, BOTH);
+   */
+  private final VertexDegree degreeType;
+
+  /**
+   * Creates an instance of this average degree evolution operator.
+   *
+   * @param degreeType the degree type to use (IN, OUT, BOTH).
+   * @param dimension the time dimension to use (VALID_TIME, TRANSACTION_TIME).
+   */
+  public DegreeRangeEvolution(VertexDegree degreeType, TimeDimension dimension) {
+    this.degreeType = Objects.requireNonNull(degreeType);
+    this.dimension = Objects.requireNonNull(dimension);
+  }
+
+  @Override
+  public DataSet<Tuple3<Long, Long, Float>> execute(TemporalGraph graph) {
+    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())
+            .reduceGroup(new GroupDegreeTreesToDegreeRange())
+            .mapPartition(new MapDegreesToInterval());
+  }
+}

gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/GroupDegreeTreesToDegreeRange.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright © 2014 - 2021 Leipzig University (Database Research Group)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.gradoop.temporal.model.impl.operators.metric.functions;
+
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+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.HashMap;
+import java.util.Map;
+
+/**
+ * A group reduce function that merges all Tuples (vId, degreeTree) to a dataset of tuples (time, aggDegree)
+ * that represents the aggregated degree value for the whole graph at the given time.
+ */
+public class GroupDegreeTreesToDegreeRange
+        implements GroupReduceFunction<Tuple2<GradoopId, TreeMap<Long, Integer>>, Tuple2<Long, Float>> {
+
+   /**
+   * Creates an instance of this group reduce function.
+   *
+   */
+  public GroupDegreeTreesToDegreeRange() {
+
+  }
+
+  @Override
+  public void reduce(Iterable<Tuple2<GradoopId, TreeMap<Long, Integer>>> iterable,
+                       Collector<Tuple2<Long, Float>> collector) throws Exception {
+
+    // 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());
+    }
+
+    // Add default times
+    timePoints.add(Long.MIN_VALUE);
+
+    for (Long timePoint : timePoints) {
+        // skip last default time
+        /*if (Long.MAX_VALUE == timePoint) {
+            continue;
+        }*/
+        // Iterate over all vertices
+      for (Map.Entry<GradoopId, TreeMap<Long, Integer>> entry : degreeTrees.entrySet()) {
+        // Make sure the vertex is registered in the current vertexDegrees capture
+        if (!vertexDegrees.containsKey(entry.getKey())) {
+          vertexDegrees.put(entry.getKey(), 0);
+        }
+
+        // Check if timestamp is in tree, if not, take the lower key
+        if (entry.getValue().containsKey(timePoint)) {
+          vertexDegrees.put(entry.getKey(), entry.getValue().get(timePoint));
+        } else {
+          Long lowerKey = entry.getValue().lowerKey(timePoint);
+          if (lowerKey != null) {
+            vertexDegrees.put(entry.getKey(), entry.getValue().get(lowerKey));
+          }
+        }
+      }
+
+      // Here, every tree with this time point is iterated. Now we need to aggregate for the current time.
+      float maxDegree = vertexDegrees.values().stream().reduce(Math::max).orElse(0).floatValue();
+      float minDegree = vertexDegrees.values().stream().reduce(Math::min).orElse(0).floatValue();
+      collector.collect(new Tuple2<>(timePoint, maxDegree - minDegree));
+    }
+  }
+}

gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/MapDegreesToInterval.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright © 2014 - 2021 Leipzig University (Database Research Group)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.gradoop.temporal.model.impl.operators.metric.functions;
+
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.util.Collector;
+
+/**
+ * A map partition function that calculates the intervals in which the values stay the same
+ * by checking if the current value is the same as the previous one
+ *
+ */
+
+public class MapDegreesToInterval implements MapPartitionFunction<Tuple2<Long, Float>, Tuple3<Long, Long, Float>> {
+  @Override
+  public void mapPartition(Iterable<Tuple2<Long, Float>> values, Collector<Tuple3<Long, Long, Float>> out) {
+
+    //set starting values to null
+    Long startTimestamp = null;
+    Long endTimestamp = null;
+    Float value = null;
+    Boolean collected = false;
+
+    //loop through each tuple
+    for (Tuple2<Long, Float> tuple : values) {
+      if (startTimestamp == null) {
+        // First element in the group
+        startTimestamp = tuple.f0;
+        endTimestamp = tuple.f0;
+        value = tuple.f1;
+      } else {
+        if (!tuple.f1.equals(value)) {
+          // Value changed, emit the current interval and start a new one
+          out.collect(new Tuple3<>(startTimestamp, tuple.f0, value));
+          startTimestamp = tuple.f0;
+          endTimestamp = tuple.f0;
+          value = tuple.f1;
+          collected = true;
+        } else {
+          // Extend the current interval
+          endTimestamp = tuple.f0;
+          collected = false;
+        }
+      }
+    }
+    //check if the latest interval was collected, if not, collect it
+    //this happens when the last interval has the value 0
+    if (!collected) {
+      out.collect(new Tuple3<>(startTimestamp, endTimestamp, value));
+    }
+  }
+}

gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/TransformDeltaToAbsoluteDegreeTree.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright © 2014 - 2021 Leipzig University (Database Research Group)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.gradoop.temporal.model.impl.operators.metric.functions;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.functions.FunctionAnnotation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.gradoop.common.model.impl.id.GradoopId;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Replaces the degree tree, that just stores the degree changes for each time, with a degree tree that
+ * stores the actual degree of the vertex at that time.
+ */
+@FunctionAnnotation.ForwardedFields("f0")
+public class TransformDeltaToAbsoluteDegreeTree
+        implements MapFunction<Tuple2<GradoopId, TreeMap<Long, Integer>>,
+        Tuple2<GradoopId, TreeMap<Long, Integer>>> {
+
+    /**
+     * To reduce object instantiations.
+     */
+  private TreeMap<Long, Integer> absoluteDegreeTree;
+
+  @Override
+  public Tuple2<GradoopId, TreeMap<Long, Integer>> map(
+        Tuple2<GradoopId, TreeMap<Long, Integer>> vIdTreeMapTuple) throws Exception {
+    // init the degree and the temporal tree
+    int degree = 0;
+    absoluteDegreeTree = new TreeMap<>();
+
+    // aggregate the degrees
+    for (Map.Entry<Long, Integer> entry : vIdTreeMapTuple.f1.entrySet()) {
+      degree += entry.getValue();
+      absoluteDegreeTree.put(entry.getKey(), degree);
+    }
+    vIdTreeMapTuple.f1 = absoluteDegreeTree;
+    return vIdTreeMapTuple;
+  }
+}