GitHub PullRequestReviewCommentEvent at 2014-05-21 14:35:09
Repository: stratosphere.github.io
Actor: rmetzger
Payload: {"comment":{"url":"https://api.github.com/repos/stratosphere/stratosphere.github.io/pulls/comments/12898954","id":12898954,"diff_hunk":"@@ -0,0 +1,358 @@\n+---\n+layout: inner_docs_v05\n+title:  \"Java Example Programs\"\n+sublinks:\n+  - {anchor: \"wordcount\", title: \"Word Count\"}\n+  - {anchor: \"page_rank\", title: \"Page Rank\"}\n+  - {anchor: \"connected_components\", title: \"Connected Components\"}\n+  - {anchor: \"relational\", title: \"Relational Query\"}\n+---\n+\n+<!--\n+<div class=\"panel panel-default\"><div class=\"panel-body\">Please note that parts of the documentation are out of sync. We are in the process of updating everything to reflect the changes of the upcoming release. If you have any questions, we are happy to help on our <a href=\"{{site.baseurl}}/project/contact/\">Mailinglist</a> or on <a href=\"https://github.com/stratosphere/stratosphere/issues\">GitHub</a>.</div></div>\n+-->\n+\n+## Java Example Programs\n+\n+<p class=\"lead\">\n+The following example programs showcase different applications of Stratosphere from simple word counting to graph algorithms.\n+The code samples illustrate the use of **[Stratosphere's Java API]({{site.baseurl}}/docs/{{site.current_stable}}/programming_guides/java.html)**. \n+\n+<p class=\"lead\">\n+The full source code of the following and more examples can be found in the **[stratosphere-java-examples](https://github.com/stratosphere/stratosphere/tree/release-{{site.current_stable}}/stratosphere-examples/stratosphere-java-examples)** module.\n+\n+<section id=\"wordcount\">\n+<div class=\"page-header\"><h2>Word Count</h2></div>\n+\n+WordCount is the \"Hello World\" of Big Data processing systems. It computes the frequency of words in a text collection. The algorithm works in two steps: First, the texts are tokenized to individual words (with optional normalization). Second, the words are grouped and counted.\n+\n+<div id=\"wordcount-java-code\">\n+{% highlight java %}\n+\n+// get input data\n+DataSet<String> text = getTextDataSet(env);\n+\n+DataSet<Tuple2<String, Integer>> counts = \n+        // split up the lines in pairs (2-tuples) containing: (word,1)\n+        text.flatMap(new Tokenizer())\n+        // group by the tuple field \"0\" and sum up tuple field \"1\"\n+        .groupBy(0)\n+        .aggregate(Aggregations.SUM, 1);\n+\n+counts.writeAsCsv(outputPath, \"\\n\", \" \");\n+\n+// User-defined functions\n+public static final class Tokenizer extends FlatMapFunction<String, Tuple2<String, Integer>> {\n+\n+    @Override\n+    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {\n+        // normalize and split the line\n+        String[] tokens = value.toLowerCase().split(\"\\\\W+\");\n+        \n+        // emit the pairs\n+        for (String token : tokens) {\n+            if (token.length() > 0) {\n+                out.collect(new Tuple2<String, Integer>(token, 1));\n+            }   \n+        }\n+    }\n+}\n+{% endhighlight %}\n+\n+</div>\n+\n+The <a href=\"https://github.com/stratosphere/stratosphere/blob/release-{{site.current_stable}}/stratosphere-examples/stratosphere-java-examples/src/main/java/eu/stratosphere/example/java/wordcount/WordCount.java\">WordCount example</a> implements the above described algorithm with input parameters: <code>&lt;text input path&gt;, &lt;output path&gt;</code>.<br>\n+As test data, any text file will do.\n+\n+</section>\n+\n+\n+<section id=\"page_rank\">\n+<div class=\"page-header\"><h2>Page Rank</h2></div>\n+\n+The PageRank algorithm computes the \"importance\" of pages in a graph defined by links, which point from one pages to another page. It is an iterative graph algorithm, which means that it repeatedly applies the same computation. In each iteration, each page distributes its current rank over all its neighbors, and compute its new rank as a taxed sum of the ranks it received from its neighbors. <br>\n+\n+In this simple example, PageRank is implemented with a bulk iteration and a fixed number of iterations.\n+\n+<div id=\"pagerank-java-code\">\n+{% highlight java %}\n+// get input data\n+DataSet<Tuple2<Long, Double>> pagesWithRanks = getPagesWithRanksDataSet(env);\n+DataSet<Tuple2<Long, Long[]>> pageLinkLists = getLinksDataSet(env);\n+\n+// set iterative data set\n+IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);\n+\n+DataSet<Tuple2<Long, Double>> newRanks = iteration\n+        // join pages with outgoing edges and distribute rank\n+        .join(pageLinkLists).where(0).equalTo(0).flatMap(new JoinVertexWithEdgesMatch())\n+        // collect and sum ranks\n+        .groupBy(0).aggregate(SUM, 1)\n+        // apply dampening factor\n+        .map(new Dampener(DAMPENING_FACTOR, numPages));\n+\n+DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(\n+        newRanks, \n+        newRanks.join(iteration).where(0).equalTo(0)\n+        // termination condition\n+        .filter(new EpsilonFilter()));\n+\n+finalPageRanks.writeAsCsv(outputPath, \"\\n\", \" \");\n+\n+// User-defined functions\n+\n+public static final class JoinVertexWithEdgesMatch \n+                    extends FlatMapFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>>, \n+                                            Tuple2<Long, Double>> {\n+\n+    @Override\n+    public void flatMap(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>> value, \n+                        Collector<Tuple2<Long, Double>> out) {\n+        Long[] neigbors = value.f1.f1;\n+        double rank = value.f0.f1;\n+        double rankToDistribute = rank / ((double) neigbors.length);\n+            \n+        for (int i = 0; i < neigbors.length; i++) {\n+            out.collect(new Tuple2<Long, Double>(neigbors[i], rankToDistribute));\n+        }\n+    }\n+}\n+\n+public static final class Dampener extends MapFunction<Tuple2<Long,Double>, Tuple2<Long,Double>> {\n+    private final double dampening, randomJump;\n+\n+    public Dampener(double dampening, double numVertices) {\n+        this.dampening = dampening;\n+        this.randomJump = (1 - dampening) / numVertices;\n+    }\n+\n+    @Override\n+    public Tuple2<Long, Double> map(Tuple2<Long, Double> value) {\n+        value.f1 = (value.f1 * dampening) + randomJump;\n+        return value;\n+    }\n+}\n+\n+public static final class EpsilonFilter \n+                    extends FilterFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>>> {\n+\n+    @Override\n+    public boolean filter(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) {\n+        return Math.abs(value.f0.f1 - value.f1.f1) > EPSILON;\n+    }\n+}\n+\n+{% endhighlight %}\n+\n+</div>\n+\n+The <a href=\"https://github.com/stratosphere/stratosphere/blob/release-{{site.current_stable}}/stratosphere-examples/stratosphere-java-examples/src/main/java/eu/stratosphere/example/java/graph/PageRankBasic.java\">PageRank program</a> implements the above example.\n+It requires the following parameters to run: <code>&lt;pages input path&gt;, &lt;links input path&gt;, &lt;output path&gt; &lt;num pages&gt;, &lt;num iterations&gt;</code>.<br>\n+<p>\n+Input files are plain text files and must be formatted as follows:\n+<ul>\n+<li>Pages represented as an (long) ID separated by new-line characters.<br> \n+    For example <code>\"1\\n2\\n12\\n42\\n63\\n\"</code> gives five pages with IDs 1, 2, 12, 42, and 63.\n+<li>Links are represented as pairs of page IDs which are separated by space characters. Links are separated by new-line characters.<br>\n+    For example <code>\"1 2\\n2 12\\n1 12\\n42 63\\n\"</code> gives four (directed) links (1)->(2), (2)->(12), (1)->(12), and (42)->(63).<br>\n+    For this simple implementation it is required that each page has at least one incoming and one outgoing link (a page can point to itself).\n+</ul>\n+\n+</section>\n+\n+<section id=\"connected_components\">\n+<div class=\"page-header\"><h2>Connected Components</h2></div>\n+\n+The Connected Components algorithm identifies components of a larger graph which are connected by assigning all vertices in the same component the same component ID. Similar to PageRank, Connected Components is an iterative algorithm. In each step, each vertex propagates its current component ID to all its neighbors. A vertex accepts the component ID from a neighbor, if it is smaller than its own component ID.<br>\n+\n+This implementation uses a delta iteration: Vertices that have not changed their component ID do not participate in the next step. This yields much better performance, because the later iterations typically deal only with a few outlier vertices.\n+\n+<div id=\"cc-java-code\">\n+\n+{% highlight java %}\n+\n+// read vertex and edge data\n+DataSet<Long> vertices = getVertexDataSet(env);\n+DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());\n+\n+// assign the initial component IDs (equal to the vertex ID)\n+DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());\n+        \n+// open a delta iteration\n+DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =\n+        verticesWithInitialId.iterateDelta(verticesWithInitialId, maxIterations, 0);\n+\n+// apply the step logic: \n+DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset()\n+        // join with the edges\n+        .join(edges).where(0).equalTo(0).with(new NeighborWithComponentIDJoin())\n+        // select the minimum neighbor component ID\n+        .groupBy(0).aggregate(Aggregations.MIN, 1)\n+        // update if the component ID of the candidate is smaller\n+        .join(iteration.getSolutionSet()).where(0).equalTo(0)\n+        .flatMap(new ComponentIdFilter());\n+\n+// close the delta iteration (delta and new workset are identical)\n+DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);\n+\n+// emit result\n+result.writeAsCsv(outputPath, \"\\n\", \" \");\n+\n+// User-defined functions\n+\n+public static final class DuplicateValue<T> extends MapFunction<T, Tuple2<T, T>> {\n+    \n+    @Override\n+    public Tuple2<T, T> map(T vertex) {\n+        return new Tuple2<T, T>(vertex, vertex);\n+    }\n+}\n+\n+public static final class UndirectEdge \n+                    extends FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {\n+    Tuple2<Long, Long> invertedEdge = new Tuple2<Long, Long>();\n+    \n+    @Override\n+    public void flatMap(Tuple2<Long, Long> edge, Collector<Tuple2<Long, Long>> out) {\n+        invertedEdge.f0 = edge.f1;\n+        invertedEdge.f1 = edge.f0;\n+        out.collect(edge);\n+        out.collect(invertedEdge);\n+    }\n+}\n+\n+public static final class NeighborWithComponentIDJoin \n+                    extends JoinFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> {\n+\n+    @Override\n+    public Tuple2<Long, Long> join(Tuple2<Long, Long> vertexWithComponent, Tuple2<Long, Long> edge) {\n+        return new Tuple2<Long, Long>(edge.f1, vertexWithComponent.f1);\n+    }\n+}\n+\n+public static final class ComponentIdFilter \n+                    extends FlatMapFunction<Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>>, \n+                                            Tuple2<Long, Long>> {\n+\n+    @Override\n+    public void flatMap(Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>> value, \n+                        Collector<Tuple2<Long, Long>> out) {\n+        if (value.f0.f1 < value.f1.f1) {\n+            out.collect(value.f0);\n+        }\n+    }\n+}\n+\n+{% endhighlight %}\n+\n+</div>\n+\n+The <a href=\"https://github.com/stratosphere/stratosphere/blob/release-{{site.current_stable}}/stratosphere-examples/stratosphere-java-examples/src/main/java/eu/stratosphere/example/java/graph/ConnectedComponents.java\">ConnectedComponents program</a> implements the above example.\n+It requires the following parameters to run: <code>&lt;vertex input path&gt;, &lt;edge input path&gt;, &lt;output path&gt; &lt;max num iterations&gt;</code>.<br>\n+<p>\n+Input files are plain text files and must be formatted as follows:\n+<ul>\n+<li>Vertices represented as IDs and separated by new-line characters.<br> \n+    For example <code>\"1\\n2\\n12\\n42\\n63\\n\"</code> gives five vertices (1), (2), (12), (42), and (63). \n+<li>Edges are represented as pairs for vertex IDs which are separated by space characters. Edges are separated by new-line characters.<br>\n+    For example <code>\"1 2\\n2 12\\n1 12\\n42 63\\n\"</code> gives four (undirected) edges (1)-(2), (2)-(12), (1)-(12), and (42)-(63).\n+</ul>\n+\n+</section>\n+\n+\n+<section id=\"relational\">\n+<div class=\"page-header\"><h2>Relational Query</h2></div>\n+\n+The Relational Query example assumes two tables, one with `orders` and the other with `lineitems` as specified by the [TPC-H decision support benchmark](http://www.tpc.org/tpch/). See below for instructions how to generate this data.<br>","path":"docs/0.5/programming_guides/examples_java.markdown","position":267,"original_position":267,"commit_id":"e8fe6a75929995855f7da2e83fcd867c7c4a19b4","original_commit_id":"e8fe6a75929995855f7da2e83fcd867c7c4a19b4","user":{"login":"rmetzger","id":89049,"avatar_url":"https://avatars.githubusercontent.com/u/89049?","gravatar_id":"89c466492a2ac4cf61e8aee4c070f54c","url":"https://api.github.com/users/rmetzger","html_url":"https://github.com/rmetzger","followers_url":"https://api.github.com/users/rmetzger/followers","following_url":"https://api.github.com/users/rmetzger/following{/other_user}","gists_url":"https://api.github.com/users/rmetzger/gists{/gist_id}","starred_url":"https://api.github.com/users/rmetzger/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rmetzger/subscriptions","organizations_url":"https://api.github.com/users/rmetzger/orgs","repos_url":"https://api.github.com/users/rmetzger/repos","events_url":"https://api.github.com/users/rmetzger/events{/privacy}","received_events_url":"https://api.github.com/users/rmetzger/received_events","type":"User","site_admin":false},"body":"\"The TPC-H benchmark is a standard benchmark in the database industry. It is used to test the performance of database systems. The benchmark includes a tool to generate sample data at different sizes and predefined queries\"\r\n(I made the experience that many people, even in the big data space don't know TPC-H)","created_at":"2014-05-21T14:35:09Z","updated_at":"2014-05-21T14:35:09Z","html_url":"https://github.com/stratosphere/stratosphere.github.io/pull/45#discussion_r12898954","pull_request_url":"https://api.github.com/repos/stratosphere/stratosphere.github.io/pulls/45","_links":{"self":{"href":"https://api.github.com/repos/stratosphere/stratosphere.github.io/pulls/comments/12898954"},"html":{"href":"https://github.com/stratosphere/stratosphere.github.io/pull/45#discussion_r12898954"},"pull_request":{"href":"https://api.github.com/repos/stratosphere/stratosphere.github.io/pulls/45"}}}}