You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2016/05/25 16:56:26 UTC
flink git commit: [FLINK-2044] [gelly] Implementation of HITS
algorithm.
Repository: flink
Updated Branches:
refs/heads/master af0f41824 -> ec38a212e
[FLINK-2044] [gelly] Implementation of HITS algorithm.
This closes #1956
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ec38a212
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ec38a212
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ec38a212
Branch: refs/heads/master
Commit: ec38a212e9ffa799f6ec20166179da25d5d240ba
Parents: af0f418
Author: gallenvara <ga...@126.com>
Authored: Mon May 2 17:58:17 2016 +0800
Committer: vasia <va...@apache.org>
Committed: Wed May 25 18:54:27 2016 +0200
----------------------------------------------------------------------
docs/apis/batch/libs/gelly.md | 18 ++
.../flink/graph/examples/data/HITSData.java | 71 +++++
.../graph/library/HITSAlgorithmITCase.java | 133 ++++++++
.../flink/graph/library/HITSAlgorithm.java | 312 +++++++++++++++++++
4 files changed, 534 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ec38a212/docs/apis/batch/libs/gelly.md
----------------------------------------------------------------------
diff --git a/docs/apis/batch/libs/gelly.md b/docs/apis/batch/libs/gelly.md
index 0790c29..0d3e594 100644
--- a/docs/apis/batch/libs/gelly.md
+++ b/docs/apis/batch/libs/gelly.md
@@ -1830,6 +1830,7 @@ Gelly has a growing collection of graph algorithms for easily analyzing large-sc
* [GSA Single Source Shortest Paths](#gsa-single-source-shortest-paths)
* [GSA Triangle Count](#gsa-triangle-count)
* [Triangle Enumerator](#triangle-enumerator)
+* [Hyperlink-Induced Topic Search](#hyperlink-induced-topic-search)
* [Summarization](#summarization)
* [Jaccard Index](#jaccard-index)
* [Local Clustering Coefficient](#local-clustering-coefficient)
@@ -2028,6 +2029,23 @@ This implementation extends the basic algorithm by computing output degrees of e
The algorithm takes a directed graph as input and outputs a `DataSet` of `Tuple3`. The Vertex ID type has to be `Comparable`.
Each `Tuple3` corresponds to a triangle, with the fields containing the IDs of the vertices forming the triangle.
+### Hyperlink-Induced Topic Search
+
+#### Overview
+[Hyperlink-Induced Topic Search](http://www.cs.cornell.edu/home/kleinber/auth.pdf) (HITS, or "Hubs and Authorities")
+computes two interdependent scores for every vertex in a directed graph. Good hubs are those which point to many
+good authorities and good authorities are those pointed to by many good hubs.
+
+#### Details
+HITS ranking relies on an iterative method converging to a stationary solution. Each vertex in the directed graph is assigned same non-negative
+hub and authority scores. Then the algorithm iteratively updates the scores until termination. Current implementation divides the iteration
+into two phases, authority scores can be computed until hub scores updating and normalising finished, hub scores can be computed until
+authority scores updating and normalising finished.
+
+#### Usage
+The algorithm takes a directed graph as input and outputs a `DataSet` of vertices, where the vertex value is a `Tuple2`
+containing the hub and authority score after maximum iterations.
+
### Summarization
#### Overview
http://git-wip-us.apache.org/repos/asf/flink/blob/ec38a212/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/data/HITSData.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/data/HITSData.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/data/HITSData.java
new file mode 100644
index 0000000..7f40856
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/data/HITSData.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.flink.graph.examples.data;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.types.NullValue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Provides the data set used for the HITS test program.
+ */
+public class HITSData {
+
+ public static final String VALUE_AFTER_10_ITERATIONS = "1,0.70710678,3.12608866E-8\n" +
+ "2,1.29486832E-8,0.70710678\n" +
+ "3,1.29486832E-8,0.49999999\n" +
+ "4,0.50000001,0.49999999\n" +
+ "5,0.50000001,3.12608866E-8\n";
+
+
+ private HITSData() {}
+
+ public static final DataSet<Vertex<Long, Double>> getVertexDataSet(ExecutionEnvironment env) {
+
+ List<Vertex<Long, Double>> vertices = new ArrayList<>();
+ vertices.add(new Vertex<>(1L, 1.0));
+ vertices.add(new Vertex<>(2L, 2.0));
+ vertices.add(new Vertex<>(3L, 3.0));
+ vertices.add(new Vertex<>(4L, 4.0));
+ vertices.add(new Vertex<>(5L, 5.0));
+
+ return env.fromCollection(vertices);
+ }
+
+ public static final DataSet<Edge<Long, NullValue>> getEdgeDataSet(ExecutionEnvironment env) {
+
+ List<Edge<Long, NullValue>> edges = new ArrayList<>();
+ edges.add(new Edge<>(2L, 1L, NullValue.getInstance()));
+ edges.add(new Edge<>(5L, 2L, NullValue.getInstance()));
+ edges.add(new Edge<>(5L, 4L, NullValue.getInstance()));
+ edges.add(new Edge<>(4L, 3L, NullValue.getInstance()));
+ edges.add(new Edge<>(4L, 2L, NullValue.getInstance()));
+ edges.add(new Edge<>(1L, 4L, NullValue.getInstance()));
+ edges.add(new Edge<>(1L, 2L, NullValue.getInstance()));
+ edges.add(new Edge<>(1L, 3L, NullValue.getInstance()));
+ edges.add(new Edge<>(3L, 5L, NullValue.getInstance()));
+
+ return env.fromCollection(edges);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ec38a212/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/library/HITSAlgorithmITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/library/HITSAlgorithmITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/library/HITSAlgorithmITCase.java
new file mode 100644
index 0000000..019b851
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/library/HITSAlgorithmITCase.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.flink.graph.library;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.examples.data.HITSData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.DoubleValue;
+import org.apache.flink.types.NullValue;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.List;
+
+@RunWith(Parameterized.class)
+public class HITSAlgorithmITCase extends MultipleProgramsTestBase{
+
+ public HITSAlgorithmITCase(TestExecutionMode mode) {
+ super(mode);
+ }
+
+ @Test
+ public void testHITSWithTenIterations() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Double, NullValue> graph = Graph.fromDataSet(
+ HITSData.getVertexDataSet(env),
+ HITSData.getEdgeDataSet(env),
+ env);
+
+ List<Vertex<Long, Tuple2<DoubleValue, DoubleValue>>> result = graph.run(new HITSAlgorithm<Long, Double, NullValue>(10)).collect();
+
+ compareWithDelta(result, 1e-7);
+ }
+
+ @Test
+ public void testHITSWithTenIterationsAndNumOfVertices() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Double, NullValue> graph = Graph.fromDataSet(
+ HITSData.getVertexDataSet(env),
+ HITSData.getEdgeDataSet(env),
+ env);
+
+ List<Vertex<Long, Tuple2<DoubleValue, DoubleValue>>> result = graph.run(new HITSAlgorithm<Long, Double, NullValue>(10, 5)).collect();
+
+ compareWithDelta(result, 1e-7);
+ }
+
+ @Test
+ public void testHITSWithConvergeThreshold() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Double, NullValue> graph = Graph.fromDataSet(
+ HITSData.getVertexDataSet(env),
+ HITSData.getEdgeDataSet(env),
+ env);
+
+ List<Vertex<Long, Tuple2<DoubleValue, DoubleValue>>> result = graph.run(new HITSAlgorithm<Long, Double, NullValue>(1e-7)).collect();
+
+ compareWithDelta(result, 1e-7);
+ }
+
+ @Test
+ public void testHITSWithConvergeThresholdAndNumOfVertices() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Double, NullValue> graph = Graph.fromDataSet(
+ HITSData.getVertexDataSet(env),
+ HITSData.getEdgeDataSet(env),
+ env);
+
+ List<Vertex<Long, Tuple2<DoubleValue, DoubleValue>>> result = graph.run(new HITSAlgorithm<Long, Double, NullValue>(1e-7, 5)).collect();
+
+ compareWithDelta(result, 1e-7);
+ }
+
+ private void compareWithDelta(List<Vertex<Long, Tuple2<DoubleValue, DoubleValue>>> result, double delta) {
+
+ String resultString = "";
+ for (Vertex<Long, Tuple2<DoubleValue, DoubleValue>> v : result) {
+ resultString += v.f0.toString() + "," + v.f1.f0.toString() + "," + v.f1.f1.toString() +"\n";
+ }
+
+ String expectedResult = HITSData.VALUE_AFTER_10_ITERATIONS;
+
+ String[] expected = expectedResult.isEmpty() ? new String[0] : expectedResult.split("\n");
+
+ String[] resultArray = resultString.isEmpty() ? new String[0] : resultString.split("\n");
+
+ Arrays.sort(expected);
+ Arrays.sort(resultArray);
+
+ for (int i = 0; i < expected.length; i++) {
+ String[] expectedFields = expected[i].split(",");
+ String[] resultFields = resultArray[i].split(",");
+
+ double expectedHub = Double.parseDouble(expectedFields[1]);
+ double resultHub = Double.parseDouble(resultFields[1]);
+
+ double expectedAuthority = Double.parseDouble(expectedFields[2]);
+ double resultAuthority = Double.parseDouble(resultFields[2]);
+
+ Assert.assertTrue("Values differ by more than the permissible delta",
+ Math.abs(expectedHub - resultHub) < delta);
+
+ Assert.assertTrue("Values differ by more than the permissible delta",
+ Math.abs(expectedAuthority - resultAuthority) < delta);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ec38a212/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/HITSAlgorithm.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/HITSAlgorithm.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/HITSAlgorithm.java
new file mode 100644
index 0000000..1ea367e
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/HITSAlgorithm.java
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.flink.graph.library;
+
+import org.apache.flink.api.common.aggregators.DoubleSumAggregator;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.GraphAlgorithm;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.ScatterGatherConfiguration;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+import org.apache.flink.types.DoubleValue;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * This is an implementation of HITS algorithm, using a scatter-gather iteration.
+ * The user can define the maximum number of iterations. HITS algorithm is determined by two parameters,
+ * hubs and authorities. A good hub represents a page that points to many other pages, and a good authority
+ * represented a page that is linked by many different hubs.
+ * Each vertex has a value of Tuple2 type, the first field is hub score and the second field is authority score.
+ * The implementation sets same score to every vertex and adds the reverse edge to every edge at the beginning.
+ * If the number of vertices of the input graph is known, it should be provided as a parameter
+ * to speed up computation. Otherwise, the algorithm will first execute a job to count the vertices.
+ * <p>
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/HITS_algorithm">HITS Algorithm</a>
+ */
+public class HITSAlgorithm<K, VV, EV> implements GraphAlgorithm<K, VV, EV, DataSet<Vertex<K, Tuple2<DoubleValue, DoubleValue>>>> {
+
+ private final static int MAXIMUMITERATION = (Integer.MAX_VALUE - 1) / 2;
+ private final static double MINIMUMTHRESHOLD = 1e-9;
+
+ private int maxIterations;
+ private long numberOfVertices;
+ private double convergeThreshold;
+
+ /**
+ * Create an instance of HITS algorithm.
+ *
+ * @param maxIterations the maximum number of iterations
+ */
+ public HITSAlgorithm(int maxIterations) {
+ this(maxIterations, MINIMUMTHRESHOLD);
+ }
+
+ /**
+ * Create an instance of HITS algorithm.
+ *
+ * @param convergeThreshold convergence threshold for sum of scores to control whether the iteration should be stopped
+ */
+ public HITSAlgorithm(double convergeThreshold) {
+ this(MAXIMUMITERATION, convergeThreshold);
+ }
+
+ /**
+ * Create an instance of HITS algorithm.
+ *
+ * @param maxIterations the maximum number of iterations
+ * @param numberOfVertices the number of vertices in the graph
+ */
+ public HITSAlgorithm(int maxIterations, long numberOfVertices) {
+ this(maxIterations, MINIMUMTHRESHOLD, numberOfVertices);
+ }
+
+ /**
+ * Create an instance of HITS algorithm.
+ *
+ * @param convergeThreshold convergence threshold for sum of scores to control whether the iteration should be stopped
+ * @param numberOfVertices the number of vertices in the graph
+ */
+ public HITSAlgorithm(double convergeThreshold, long numberOfVertices) {
+ this(MAXIMUMITERATION, convergeThreshold, numberOfVertices);
+ }
+
+ /**
+ * Creates an instance of HITS algorithm.
+ *
+ * @param maxIterations the maximum number of iterations
+ * @param convergeThreshold convergence threshold for sum of scores to control whether the iteration should be stopped
+ */
+ public HITSAlgorithm(int maxIterations, double convergeThreshold) {
+ Preconditions.checkArgument(maxIterations > 0, "Number of iterations must be greater than zero.");
+ Preconditions.checkArgument(convergeThreshold > 0.0, "Convergence threshold must be greater than zero.");
+ this.maxIterations = maxIterations * 2 + 1;
+ this.convergeThreshold = convergeThreshold;
+ }
+
+ /**
+ * Creates an instance of HITS algorithm.
+ *
+ * @param maxIterations the maximum number of iterations
+ * @param convergeThreshold convergence threshold for sum of scores to control whether the iteration should be stopped
+ * @param numberOfVertices the number of vertices in the graph
+ */
+ public HITSAlgorithm(int maxIterations, double convergeThreshold, long numberOfVertices) {
+ this(maxIterations, convergeThreshold);
+ Preconditions.checkArgument(numberOfVertices > 0, "Number of vertices must be greater than zero.");
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ @Override
+ public DataSet<Vertex<K, Tuple2<DoubleValue, DoubleValue>>> run(Graph<K, VV, EV> graph) throws Exception {
+
+ if (numberOfVertices == 0) {
+ numberOfVertices = graph.numberOfVertices();
+ }
+
+ Graph<K, Tuple2<DoubleValue, DoubleValue>, Boolean> newGraph = graph
+ .mapEdges(new AuthorityEdgeMapper<K, EV>())
+ .union(graph.reverse().mapEdges(new HubEdgeMapper<K, EV>()))
+ .mapVertices(new VertexInitMapper<K, VV>());
+
+ ScatterGatherConfiguration parameter = new ScatterGatherConfiguration();
+ parameter.setDirection(EdgeDirection.OUT);
+ parameter.registerAggregator("updatedValueSum", new DoubleSumAggregator());
+ parameter.registerAggregator("authorityValueSum", new DoubleSumAggregator());
+ parameter.registerAggregator("diffValueSum", new DoubleSumAggregator());
+
+ return newGraph
+ .runScatterGatherIteration(new VertexUpdate<K>(maxIterations, convergeThreshold, numberOfVertices),
+ new MessageUpdate<K>(maxIterations), maxIterations, parameter)
+ .getVertices();
+ }
+
+ /**
+ * Function that updates the value of a vertex by summing up the partial
+ * values from all messages and normalize the value.
+ */
+ @SuppressWarnings("serial")
+ public static final class VertexUpdate<K> extends VertexUpdateFunction<K, Tuple2<DoubleValue, DoubleValue>, Double> {
+ private int maxIteration;
+ private double convergeThreshold;
+ private long numberOfVertices;
+ private DoubleSumAggregator updatedValueSumAggregator;
+ private DoubleSumAggregator authoritySumAggregator;
+ private DoubleSumAggregator diffSumAggregator;
+
+ public VertexUpdate(int maxIteration, double convergeThreshold, long numberOfVertices) {
+ this.maxIteration = maxIteration;
+ this.convergeThreshold = convergeThreshold;
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ @Override
+ public void preSuperstep() {
+ updatedValueSumAggregator = getIterationAggregator("updatedValueSum");
+ authoritySumAggregator = getIterationAggregator("authorityValueSum");
+ diffSumAggregator = getIterationAggregator("diffValueSum");
+ }
+
+ @Override
+ public void updateVertex(Vertex<K, Tuple2<DoubleValue, DoubleValue>> vertex, MessageIterator<Double> inMessages) {
+ double updateValue = 0;
+
+ for (double element : inMessages) {
+ if (getSuperstepNumber() == maxIteration) {
+ updateValue = element;
+ break;
+ }
+ updateValue += element;
+ }
+ updatedValueSumAggregator.aggregate(Math.pow(updateValue, 2));
+
+ // in the first iteration, no aggregation to call, init sum with value of vertex
+ double iterationValueSum = 1.0;
+
+ DoubleValue newHubValue = vertex.getValue().f0;
+ DoubleValue newAuthorityValue = vertex.getValue().f1;
+
+ if (getSuperstepNumber() > 1) {
+ iterationValueSum = Math.sqrt(((DoubleValue) getPreviousIterationAggregate("updatedValueSum")).getValue());
+ }
+ if (getSuperstepNumber() < maxIteration) {
+ if (getSuperstepNumber() % 2 == 1) {
+
+ //in the first iteration, the diff is the authority value of each vertex
+ double previousAuthAverage = 1.0;
+ double diffValueSum = 1.0 * numberOfVertices;
+ if (getSuperstepNumber() > 1) {
+ previousAuthAverage = ((DoubleValue) getPreviousIterationAggregate("authorityValueSum")).getValue() / numberOfVertices;
+ diffValueSum = ((DoubleValue) getPreviousIterationAggregate("diffValueSum")).getValue();
+ }
+ authoritySumAggregator.aggregate(previousAuthAverage);
+
+ if (diffValueSum > convergeThreshold) {
+ newHubValue.setValue(newHubValue.getValue() / iterationValueSum);
+ newAuthorityValue.setValue(updateValue);
+ } else {
+
+ //scores are converged and stop iteration
+ maxIteration = getSuperstepNumber();
+ newHubValue.setValue(newHubValue.getValue() / iterationValueSum);
+ }
+ } else {
+ newHubValue.setValue(updateValue);
+ newAuthorityValue.setValue(newAuthorityValue.getValue() / iterationValueSum);
+ authoritySumAggregator.aggregate(newAuthorityValue.getValue());
+ double previousAuthAverage = ((DoubleValue) getPreviousIterationAggregate("authorityValueSum")).getValue() / numberOfVertices;
+
+ // count the diff value of sum of authority scores
+ diffSumAggregator.aggregate((previousAuthAverage - newAuthorityValue.getValue()));
+ }
+ setNewVertexValue(new Tuple2<>(newHubValue, newAuthorityValue));
+ } else if (getSuperstepNumber() == maxIteration) {
+
+ //final iteration to normalize hub score
+ newHubValue.setValue(newHubValue.getValue() / iterationValueSum);
+ setNewVertexValue(new Tuple2<>(newHubValue, newAuthorityValue));
+ }
+ }
+ }
+
+ /**
+ * Distributes the value of a vertex among all neighbor vertices and sum all the
+ * value in every superstep.
+ */
+ @SuppressWarnings("serial")
+ public static final class MessageUpdate<K> extends MessagingFunction<K, Tuple2<DoubleValue, DoubleValue>, Double, Boolean> {
+ private int maxIteration;
+
+ public MessageUpdate(int maxIteration) {
+ this.maxIteration = maxIteration;
+ }
+
+ @Override
+ public void sendMessages(Vertex<K, Tuple2<DoubleValue, DoubleValue>> vertex) {
+
+ // in the first iteration, no aggregation to call, init sum with value of vertex
+ double iterationValueSum = 1.0;
+
+ if (getSuperstepNumber() > 1) {
+ iterationValueSum = Math.sqrt(((DoubleValue) getPreviousIterationAggregate("updatedValueSum")).getValue());
+ }
+ for (Edge<K, Boolean> edge : getEdges()) {
+ if (getSuperstepNumber() != maxIteration) {
+ if (getSuperstepNumber() % 2 == 1) {
+ if (edge.getValue()) {
+ sendMessageTo(edge.getTarget(), vertex.getValue().f0.getValue() / iterationValueSum);
+ }
+ } else {
+ if (!edge.getValue()) {
+ sendMessageTo(edge.getTarget(), vertex.getValue().f1.getValue() / iterationValueSum);
+ }
+ }
+ } else {
+ if (!edge.getValue()) {
+ sendMessageTo(edge.getTarget(), iterationValueSum);
+ }
+ }
+ }
+ }
+ }
+
+ public static class VertexInitMapper<K, VV> implements MapFunction<Vertex<K, VV>, Tuple2<DoubleValue, DoubleValue>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Tuple2<DoubleValue, DoubleValue> initVertexValue = new Tuple2<>(new DoubleValue(1.0), new DoubleValue(1.0));
+
+ public Tuple2<DoubleValue, DoubleValue> map(Vertex<K, VV> value) {
+
+ //init hub and authority value of each vertex
+ return initVertexValue;
+ }
+ }
+
+ public static class AuthorityEdgeMapper<K, EV> implements MapFunction<Edge<K, EV>, Boolean> {
+
+ private static final long serialVersionUID = 1L;
+
+ public Boolean map(Edge<K, EV> edge) {
+
+ // mark edge as true for authority updating
+ return true;
+ }
+ }
+
+ public static class HubEdgeMapper<K, EV> implements MapFunction<Edge<K, EV>, Boolean> {
+
+ private static final long serialVersionUID = 1L;
+
+ public Boolean map(Edge<K, EV> edge) {
+
+ // mark edge as false for hub updating
+ return false;
+ }
+ }
+}
+