diff options
author | stayhf <ubuntu@ip-10-170-187-165.us-west-1.compute.internal> | 2013-08-02 02:02:31 +0000 |
---|---|---|
committer | stayhf <chutong88@gmail.com> | 2013-08-03 06:01:16 +0000 |
commit | a6826373016f415169316024373c604089d2be31 (patch) | |
tree | 91f203c32fd5d721a3d0c5296f1446171c629eae /examples | |
parent | cad48edb70c58a48f66851e963f51db712ab7863 (diff) | |
download | spark-a6826373016f415169316024373c604089d2be31.tar.gz spark-a6826373016f415169316024373c604089d2be31.tar.bz2 spark-a6826373016f415169316024373c604089d2be31.zip |
Simple PageRank algorithm implementation in Java for SPARK-760
Diffstat (limited to 'examples')
-rw-r--r-- | examples/src/main/java/spark/examples/JavaPageRank.java | 129 |
1 files changed, 129 insertions, 0 deletions
diff --git a/examples/src/main/java/spark/examples/JavaPageRank.java b/examples/src/main/java/spark/examples/JavaPageRank.java new file mode 100644 index 0000000000..c491c86a6f --- /dev/null +++ b/examples/src/main/java/spark/examples/JavaPageRank.java @@ -0,0 +1,129 @@ +/* + * 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 spark.examples; + +import scala.Tuple2; +import spark.api.java.JavaPairRDD; +import spark.api.java.JavaRDD; +import spark.api.java.JavaSparkContext; +import spark.api.java.function.FlatMapFunction; +import spark.api.java.function.Function; +import spark.api.java.function.PairFunction; + +import java.util.List; +import java.util.ArrayList; + +/** + * Computes the PageRank of URLs from an input file. Input file should + * be in format of: + * URL neighbor URL + * URL neighbor URL + * URL neighbor URL + * ... + * where URL and their neighbors are separated by space(s). + */ +public class JavaPageRank { + private static double sum(List<Double> numbers) { + double out = 0.0; + for (double number : numbers) { + out += number; + } + return out; + } + + public static double round(double value, int places) { + if (places < 0) throw new IllegalArgumentException(); + + long factor = (long) Math.pow(10, places); + value = value * factor; + long tmp = Math.round(value); + return (double) tmp / factor; + } + + public static void main(String[] args) throws Exception { + if (args.length < 3) { + System.err.println("Usage: JavaPageRank <master> <file> <number_of_iterations>"); + System.exit(1); + } + + JavaSparkContext ctx = new JavaSparkContext(args[0], "JavaPageRank", + System.getenv("SPARK_HOME"), System.getenv("SPARK_EXAMPLES_JAR")); + + // Loads in input file. It should be in format of: + // URL neighbor URL + // URL neighbor URL + // URL neighbor URL + // ... + JavaRDD<String> lines = ctx.textFile(args[1], 1).cache(); + + // Loads all URLs from input file and initialize their neighbors. + JavaPairRDD<String, List<String>> links = lines.map(new PairFunction<String, String, String>() { + @Override + public Tuple2<String, String> call(String s) { + return new Tuple2<String, String>(s.split("\\s+")[0], s.split("\\s+")[1]); + } + }).distinct().groupByKey(); + + // Loads all URLs with other URL(s) link to from input file and initialize ranks of them to one. + JavaPairRDD<String, Double> ranks = lines.map(new PairFunction<String, String, Double>() { + @Override + public Tuple2<String, Double> call(String s) { + return new Tuple2<String, Double>(s.split("\\s+")[1], 1.0); + } + }).distinct(); + + + // Calculates and updates URL ranks continuously using PageRank algorithm. + for (int current = 0; current < Integer.parseInt(args[2]); current++) { + // Calculates URL contributions to the rank of other URLs. + JavaPairRDD<String, Double> contribs = links.join(ranks).values() + .flatMap(new FlatMapFunction<Tuple2<List<String>, Double>, Tuple2<String, Double>>() { + @Override + public Iterable<Tuple2<String, Double>> call(Tuple2<List<String>, Double> s) { + List<Tuple2<String, Double>> results = new ArrayList<Tuple2<String, Double>>(); + for (String n : s._1) { + results.add(new Tuple2<String, Double>(n, s._2 / s._1.size())); + } + + return results; + } + }).map(new PairFunction<Tuple2<String, Double>, String, Double>() { + @Override + public Tuple2<String, Double> call(Tuple2<String, Double> s) { + return s; + } + }); + + // Re-calculates URL ranks based on neighbor contributions. + ranks = contribs.groupByKey().mapValues(new Function<List<Double>, Double>() { + @Override + public Double call(List<Double> cs) throws Exception { + return 0.15 + sum(cs) * 0.85; + } + }); + } + + // Collects all URL ranks and dump them to console. + List<Tuple2<String, Double>> output = ranks.collect(); + for (Tuple2 tuple : output) { + System.out.println(tuple._1 + " has rank: " + round((Double)(tuple._2), 2) + "."); + } + + System.exit(0); + } +} |