diff --git a/gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/ExtractAllTimePointsReduce.java b/gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/ExtractAllTimePointsReduce.java new file mode 100644 index 000000000000..ece67f4a2efd --- /dev/null +++ b/gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/ExtractAllTimePointsReduce.java @@ -0,0 +1,49 @@ +/* + * 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.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.SortedSet; +import java.util.TreeMap; +import java.util.TreeSet; + +/** + * Reduce function to extract all timestamps where the degree of a vertex changes. + */ +public class ExtractAllTimePointsReduce implements GroupReduceFunction>, Tuple1> { + + public ExtractAllTimePointsReduce() { + } + + @Override + public void reduce(Iterable>> iterable, Collector> collector) throws Exception { + SortedSet timePoints = new TreeSet<>(); + + for (Tuple2> tuple : iterable) { + timePoints.addAll(tuple.f1.keySet()); + } + + for (Long timePoint: timePoints) { + collector.collect(new Tuple1<>(timePoint)); + } + + } +} diff --git a/gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/TransformDeltaToAbsoluteDegreeTree.java b/gradoop-temporal/src/main/java/org/gradoop/temporal/model/impl/operators/metric/functions/TransformDeltaToAbsoluteDegreeTree.java new file mode 100644 index 000000000000..85b8c1fadb2d --- /dev/null +++ b/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>> { + + /** + * To reduce object instantiations. + */ + private TreeMap absoluteDegreeTree; + + @Override + public Tuple2> map( + Tuple2> vIdTreeMapTuple) throws Exception { + // init the degree and the temporal tree + int degree = 0; + absoluteDegreeTree = new TreeMap<>(); + + // aggregate the degrees + for (Map.Entry entry : vIdTreeMapTuple.f1.entrySet()) { + degree += entry.getValue(); + absoluteDegreeTree.put(entry.getKey(), degree); + } + vIdTreeMapTuple.f1 = absoluteDegreeTree; + return vIdTreeMapTuple; + } +} \ No newline at end of file