You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2014/08/03 14:47:32 UTC
svn commit: r1615403 [2/3] - in /incubator/flink: build.sh site/build.sh
site/docs/0.6-SNAPSHOT/internal_general_arch.html
site/docs/0.6-SNAPSHOT/java_api_guide.html
site/docs/0.6-SNAPSHOT/java_api_transformations.html
Modified: incubator/flink/site/docs/0.6-SNAPSHOT/java_api_guide.html
URL: http://svn.apache.org/viewvc/incubator/flink/site/docs/0.6-SNAPSHOT/java_api_guide.html?rev=1615403&r1=1615402&r2=1615403&view=diff
==============================================================================
--- incubator/flink/site/docs/0.6-SNAPSHOT/java_api_guide.html (original)
+++ incubator/flink/site/docs/0.6-SNAPSHOT/java_api_guide.html Sun Aug 3 12:47:32 2014
@@ -156,9 +156,6 @@
<ul>
<li>
-<a href="#java-api">Java API</a>
-<ul>
-<li>
<a href="#introduction">Introduction</a>
</li>
<li>
@@ -174,132 +171,36 @@
<a href="#lazy-evaluation">Lazy Evaluation</a>
</li>
<li>
-<a href="#data-types">Data Types</a>
-<ul>
-<li>
-<ul>
-<li>
-<a href="#regular-types">Regular Types</a>
+<a href="#transformations">Transformations</a>
</li>
<li>
-<a href="#tuples">Tuples</a>
+<a href="#defining-keys">Defining Keys</a>
</li>
<li>
-<a href="#values">Values</a>
-</li>
-<li>
-<a href="#hadoop-writables">Hadoop Writables</a>
+<a href="#functions">Functions</a>
</li>
<li>
-<a href="#type-erasure-&-type-inferrence">Type Erasure & Type Inferrence</a>
-</li>
-</ul>
-</li>
-</ul>
-</li>
-<li>
-<a href="#data-transformations">Data Transformations</a>
-<ul>
-<li>
-<a href="#map">Map</a>
-</li>
-<li>
-<a href="#flatmap">FlatMap</a>
-</li>
-<li>
-<a href="#filter">Filter</a>
-</li>
-<li>
-<a href="#project-(tuple-datasets-only)">Project (Tuple DataSets only)</a>
-</li>
-<li>
-<a href="#transformations-on-grouped-dataset">Transformations on grouped DataSet</a>
-</li>
-<li>
-<a href="#reduce-on-grouped-dataset">Reduce on grouped DataSet</a>
-<ul>
-<li>
-<a href="#reduce-on-dataset-grouped-by-keyselector-function">Reduce on DataSet grouped by KeySelector Function</a>
-</li>
-<li>
-<a href="#reduce-on-dataset-grouped-by-field-position-keys-(tuple-datasets-only)">Reduce on DataSet grouped by Field Position Keys (Tuple DataSets only)</a>
-</li>
-</ul>
-</li>
-<li>
-<a href="#groupreduce-on-grouped-dataset">GroupReduce on grouped DataSet</a>
+<a href="#data-types">Data Types</a>
<ul>
<li>
-<a href="#groupreduce-on-dataset-grouped-by-field-position-keys-(tuple-datasets-only)">GroupReduce on DataSet grouped by Field Position Keys (Tuple DataSets only)</a>
-</li>
-<li>
-<a href="#groupreduce-on-dataset-grouped-by-keyselector-function">GroupReduce on DataSet grouped by KeySelector Function</a>
-</li>
-<li>
-<a href="#groupreduce-on-sorted-groups-(tuple-datasets-only)">GroupReduce on sorted groups (Tuple DataSets only)</a>
-</li>
-<li>
-<a href="#combinable-groupreducefunctions">Combinable GroupReduceFunctions</a>
-</li>
-</ul>
-</li>
-<li>
-<a href="#aggregate-on-grouped-tuple-dataset">Aggregate on grouped Tuple DataSet</a>
-</li>
-<li>
-<a href="#reduce-on-full-dataset">Reduce on full DataSet</a>
-</li>
-<li>
-<a href="#groupreduce-on-full-dataset">GroupReduce on full DataSet</a>
-</li>
-<li>
-<a href="#aggregate-on-full-tuple-dataset">Aggregate on full Tuple DataSet</a>
-</li>
-<li>
-<a href="#join">Join</a>
<ul>
<li>
-<a href="#default-join-(join-into-tuple2)">Default Join (Join into Tuple2)</a>
-</li>
-<li>
-<a href="#join-with-joinfunction">Join with JoinFunction</a>
-</li>
-<li>
-<a href="#join-with-projection">Join with Projection</a>
-</li>
-<li>
-<a href="#join-with-dataset-size-hint">Join with DataSet Size Hint</a>
-</li>
-</ul>
-</li>
-<li>
-<a href="#cross">Cross</a>
-<ul>
-<li>
-<a href="#cross-with-user-defined-function">Cross with User-Defined Function</a>
+<a href="#regular-types">Regular Types</a>
</li>
<li>
-<a href="#cross-with-projection">Cross with Projection</a>
+<a href="#tuples">Tuples</a>
</li>
<li>
-<a href="#cross-with-dataset-size-hint">Cross with DataSet Size Hint</a>
-</li>
-</ul>
+<a href="#values">Values</a>
</li>
<li>
-<a href="#cogroup">CoGroup</a>
-<ul>
-<li>
-<a href="#cogroup-on-datasets-grouped-by-field-position-keys-(tuple-datasets-only)">CoGroup on DataSets grouped by Field Position Keys (Tuple DataSets only)</a>
+<a href="#hadoop-writables">Hadoop Writables</a>
</li>
<li>
-<a href="#cogroup-on-datasets-grouped-by-key-selector-function">CoGroup on DataSets grouped by Key Selector Function</a>
+<a href="#type-erasure-&-type-inferrence">Type Erasure & Type Inferrence</a>
</li>
</ul>
</li>
-<li>
-<a href="#union">Union</a>
-</li>
</ul>
</li>
<li>
@@ -365,15 +266,9 @@
<a href="#execution-plans">Execution Plans</a>
</li>
</ul>
-</li>
-</ul>
-
- <p><section id="top"></section></p>
-<h1 id="java-api">Java API</h1>
-
-<p><section id="introduction"></p>
+ <p><section id="top"></p>
<h2 id="introduction">Introduction</h2>
@@ -412,7 +307,7 @@
<span class="n">env</span><span class="o">.</span><span class="na">execute</span><span class="o">(</span><span class="s">"Word Count Example"</span><span class="o">);</span>
<span class="o">}</span>
- <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kd">class</span> <span class="nc">LineSplitter</span> <span class="kd">extends</span> <span class="n">FlatMapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">LineSplitter</span> <span class="kd">implements</span> <span class="n">FlatMapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="o">{</span>
<span class="nd">@Override</span>
<span class="kd">public</span> <span class="kt">void</span> <span class="nf">flatMap</span><span class="o">(</span><span class="n">String</span> <span class="n">line</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
<span class="k">for</span> <span class="o">(</span><span class="n">String</span> <span class="n">word</span> <span class="o">:</span> <span class="n">line</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span> <span class="o">{</span>
@@ -430,7 +325,7 @@
<p>To write programs with Flink, you need to include Flinkâs Java API library in your project.</p>
-<p>The simplest way to do this is to use the <a href="java_api_quickstart.html">quickstart scripts</a>. They create a blank project from a template (a Maven Archetype), which sets up everything for you. To manually create the project, you can use the archetype and create a project by calling:</p>
+<p>The simplest way to do this is to use the <a href="/java_api_quickstart.html">quickstart scripts</a>. They create a blank project from a template (a Maven Archetype), which sets up everything for you. To manually create the project, you can use the archetype and create a project by calling:</p>
<div class="highlight"><pre><code class="language-bash" data-lang="bash">mvn archetype:generate /
-DarchetypeGroupId<span class="o">=</span>org.apache.flink/
-DarchetypeArtifactId<span class="o">=</span>flink-quickstart-java /
@@ -448,10 +343,13 @@
<span class="nt"><version></span>0.5.2<span class="nt"></version></span>
<span class="nt"></dependency></span>
</code></pre></div>
-<p>In order to link against the latest SNAPSHOT versions of the code, please follow <a href="/downloads.html/#nightly">this guide</a>.</p>
+<p>If you are using Flink together with Hadoop, the version of the dependency may vary depending on the version of Hadoop (or more specifically, HDFS) that you want to use Flink with.
+Please refer to the <a href="/downloads.html">downloads page</a> for a list of available versions, and instructions on how to link with custom versions of Hadoop.</p>
+
+<p>In order to link against the latest SNAPSHOT versions of the code, please follow <a href="/downloads.html#nightly">this guide</a>.</p>
<p>The <em>flink-clients</em> dependency is only necessary to invoke the Flink program locally (for example to run it standalone for testing and debugging).
-If you intend to only export the program as a JAR file and <a href="cluster_execution.html">run it on a cluster</a>, you can skip that dependency.</p>
+If you intend to only export the program as a JAR file and <a href="/cluster_execution.html">run it on a cluster</a>, you can skip that dependency.</p>
<p><a href="#top">Back to top</a></p>
@@ -471,7 +369,7 @@ programs with a <code>main()</code> meth
</ol>
<p>We will now give an overview of each of those steps but please refer
-to the respective sections for more details. Note that all <a href=https://github.com/apache/incubator-flink/blob/master//flink-java/src/main/java/org/apache/flink/api/java>core classes of the Java API</a> are found in the package <code>org.apache.flinkapi.java</code>.</p>
+to the respective sections for more details. Note that all <a href=https://github.com/apache/incubator-flink/blob/master//flink-java/src/main/java/org/apache/flink/api/java>core classes of the Java API</a> are found in the package <code>org.apache.flink.api.java</code>.</p>
<p>The <code>ExecutionEnvironment</code> is the basis for all Flink programs. You can
obtain one using these static methods on class <code>ExecutionEnvironment</code>:</p>
@@ -487,8 +385,8 @@ obtain one using these static methods on
will do the right thing depending on the context: if you are executing
your program inside an IDE or as a regular Java program it will create
a local environment that will execute your program on your local machine. If
-you created a JAR file from you program, and invoke it through the <a href="cli.html">command line</a>
-or the <a href="web_client.html">web interface</a>,
+you created a JAR file from you program, and invoke it through the <a href="/cli.html">command line</a>
+or the <a href="/web_client.html">web interface</a>,
the Flink cluster manager will
execute your main method and <code>getExecutionEnvironment()</code> will return
an execution environment for executing your program on a cluster.</p>
@@ -555,7 +453,231 @@ how you created the execution environmen
<p>All Flink programs are executed lazily: When the program's main method is executed, the data loading and transformations do not happen directly. Rather, each operation is created and added to the program's plan. The operations are actually executed when one of the <code>execute()</code> methods is invoked on the ExecutionEnvironment object. Whether the program is executed locally or on a cluster depends on the environment of the program.</p>
-<p>The lazy evaluation lets you construct sophisticated programs that Flink executes as one holistically planned unit.</p>
+<p>The lazy evaluation lets you construct sophisticated programs that Flink executes as one holistically planned unit.
+<a href="#top">Back to top</a></p>
+
+<p><section id="transformations"></p>
+
+<h2 id="transformations">Transformations</h2>
+
+<p>Data transformations transform one or more DataSets into a new DataSet. Programs can combine multiple transformations into
+sophisticated assemblies.</p>
+
+<p>This section gives a brief overview of the available transformations. The <a href="/java_api_transformations.html">transformations documentation</a>
+has full description of all transformations with examples.</p>
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-center" style="width: 20%">Transformation</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+
+ <tbody>
+ <tr>
+ <td><strong>Map</strong></td>
+ <td>
+ <p>Takes one element and produces one element.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="n">MapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>()</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">map</span><span class="o">(</span><span class="n">String</span> <span class="n">value</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">Integer</span><span class="o">.</span><span class="na">parseInt</span><span class="o">(</span><span class="n">value</span><span class="o">);</span> <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>FlatMap</strong></td>
+ <td>
+ <p>Takes one element and produces zero, one, or more elements.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">flatMap</span><span class="o">(</span><span class="k">new</span> <span class="n">FlatMapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="nf">flatMap</span><span class="o">(</span><span class="n">String</span> <span class="n">value</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
+ <span class="k">for</span> <span class="o">(</span><span class="n">String</span> <span class="n">s</span> <span class="o">:</span> <span class="n">value</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span> <span class="o">{</span>
+ <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">s</span><span class="o">);</span>
+ <span class="o">}</span>
+ <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Filter</strong></td>
+ <td>
+ <p>Evaluates a boolean function for each element and retains those for which the function returns *true*.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">filter</span><span class="o">(</span><span class="k">new</span> <span class="n">FilterFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">>()</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">boolean</span> <span class="nf">filter</span><span class="o">(</span><span class="n">Integer</span> <span class="n">value</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">value</span> <span class="o">></span> <span class="mi">1000</span><span class="o">;</span> <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Reduce</strong></td>
+ <td>
+ <p>Combines a group of elements into a single element by repeatedly combining two elements into one.
+ Reduce may be applied on a full data set, or on a grouped data set.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="k">new</span> <span class="n">ReduceFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Integer</span> <span class="n">a</span><span class="o">,</span> <span class="n">Integer</span> <span class="n">b</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">a</span> <span class="o">+</span> <span class="n">b</span><span class="o">;</span> <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>ReduceGroup</strong></td>
+ <td>
+ <p>Combines a group of elements into one or more elements. ReduceGroup may be applied on a full data set, or on a grouped data set.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">reduceGroup</span><span class="o">(</span><span class="k">new</span> <span class="n">GroupReduceFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="nf">reduceGroup</span><span class="o">(</span><span class="n">Iterable</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">values</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kt">int</span> <span class="n">prefixSum</span> <span class="o">=</span> <span class="mi">0</span><span class="o">;</span>
+ <span class="k">for</span> <span class="o">(</span><span class="n">Integer</span> <span class="n">i</span> <span class="o">:</span> <span class="n">values</span><span class="o">)</span> <span class="o">{</span>
+ <span class="n">prefixSum</span> <span class="o">+=</span> <span class="n">i</span><span class="o">;</span>
+ <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">prefixSum</span><span class="o">);</span>
+ <span class="o">}</span>
+ <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Aggregate</strong></td>
+ <td>
+ <p>Aggregates a group of values into a single value. Aggregation functions can be thought of as built-in reduce functions. Aggregate may be applied on a full data set, or on a grouped data set.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span><span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="n">SUM</span><span class="o">,</span> <span class="mi">0</span><span class="o">).</span><span class="na">and</span><span class="o">(</span><span class="n">MIN</span><span class="o">,</span> <span class="mi">2</span><span class="o">);</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>ReduceGroup</strong></td>
+ <td>
+ <p>Combines a group of elements into one or more elements. ReduceGroup may be applied on a full data set, or on a grouped data set.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data</span><span class="o">.</span><span class="na">reduceGroup</span><span class="o">(</span><span class="k">new</span> <span class="n">GroupReduceFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="nf">reduceGroup</span><span class="o">(</span><span class="n">Iterable</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">values</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kt">int</span> <span class="n">prefixSum</span> <span class="o">=</span> <span class="mi">0</span><span class="o">;</span>
+ <span class="k">for</span> <span class="o">(</span><span class="n">Integer</span> <span class="n">i</span> <span class="o">:</span> <span class="n">values</span><span class="o">)</span> <span class="o">{</span>
+ <span class="n">prefixSum</span> <span class="o">+=</span> <span class="n">i</span><span class="o">;</span>
+ <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">prefixSum</span><span class="o">);</span>
+ <span class="o">}</span>
+ <span class="o">}</span>
+<span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Aggregate</strong></td>
+ <td>
+ <p>Aggregates a group of values into a single value. Aggregation functions can be thought of as built-in reduce functions. Aggregate may be applied on a full data set, or on a grouped data set.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span><span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="n">SUM</span><span class="o">,</span> <span class="mi">0</span><span class="o">).</span><span class="na">and</span><span class="o">(</span><span class="n">MIN</span><span class="o">,</span> <span class="mi">2</span><span class="o">);</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ </tr>
+ <td><strong>Join</strong></td>
+ <td>
+ <p>Joins two data sets by creating all pairs of elements that are equal on their keys. Optionally uses a JoinFunction to turn the pair of elements into a single element. See [keys](#keys) on how to define join keys.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">result</span> <span class="o">=</span> <span class="n">input1</span><span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">input2</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">where</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span> <span class="c1">// key of the first input (tuple field 0)</span>
+ <span class="o">.</span><span class="na">equalTo</span><span class="o">(</span><span class="mi">1</span><span class="o">);</span> <span class="c1">// key of the second input (tuple field 1)</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>CoGroup</strong></td>
+ <td>
+ <p>The two-dimensional variant of the reduce operation. Groups each input on one or more fields and then joins the groups. The transformation function is called per pair of groups. See [keys](#keys) on how to define coGroup keys.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">data1</span><span class="o">.</span><span class="na">coGroup</span><span class="o">(</span><span class="n">data2</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">where</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">equalTo</span><span class="o">(</span><span class="mi">1</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">with</span><span class="o">(</span><span class="k">new</span> <span class="n">CoGroupFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="nf">coGroup</span><span class="o">(</span><span class="n">Iterable</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">in1</span><span class="o">,</span> <span class="n">Iterable</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">in2</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
+ <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(...);</span>
+ <span class="o">}</span>
+ <span class="o">});</span></code></pre></div>
+
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Cross</strong></td>
+ <td>
+ <p>Builds the cartesian product (cross product) of two inputs, creating all pairs of elements. Optionally uses a CrossFunction to turn the pair of elements into a single element</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">data1</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">data2</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">result</span> <span class="o">=</span> <span class="n">data1</span><span class="o">.</span><span class="na">cross</span><span class="o">(</span><span class="n">data2</span><span class="o">);</span></code></pre></div>
+
+ </td>
+ </tr>
+ <tr>
+ <td><strong>Union</strong></td>
+ <td>
+ <p>Produces the union of two data sets. This operation happens implicitly if more than one data set is used for a specific function input.</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">data1</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">data2</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">result</span> <span class="o">=</span> <span class="n">data1</span><span class="o">.</span><span class="na">union</span><span class="o">(</span><span class="n">data2</span><span class="o">);</span></code></pre></div>
+
+ </td>
+ </tr>
+ </tbody>
+</table>
+
+<hr>
+
+<p>The following transformations are available on data sets of Tuples:</p>
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-center" stype="width: 20%">Transformation</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><strong>Project</strong></td>
+ <td>
+ <p>Selects a subset of fields from the tuples</p>
+
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">in</span> <span class="o">=</span> <span class="c1">// [...]</span>
+<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="n">out</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">project</span><span class="o">(</span><span class="mi">2</span><span class="o">,</span><span class="mi">0</span><span class="o">).</span><span class="na">types</span><span class="o">(</span><span class="n">String</span><span class="o">.</span><span class="na">class</span><span class="o">,</span> <span class="n">Integer</span><span class="o">.</span><span class="na">class</span><span class="o">);</span></code></pre></div>
+
+ </td>
+ </tr>
+ </tbody>
+</table>
+
+<p><a href="#top">Back to top</a></p>
+
+<p><section id="keys"></p>
+
+<h2 id="defining-keys">Defining Keys</h2>
+
+<p><a href="#top">Back to top</a></p>
+
+<p><section id="functions"></p>
+
+<h2 id="functions">Functions</h2>
+
+<p><a href="#top">Back to top</a></p>
<p><section id="types"></p>
@@ -669,593 +791,6 @@ than one position to use composite keys
<p><a href="#top">Back to top</a></p>
-<p><section id="transformations"></p>
-
-<h2 id="data-transformations">Data Transformations</h2>
-
-<p>A data transformation transforms one or more <code>DataSet</code>s into a new <code>DataSet</code>. Advanced data analysis programs can be assembled by chaining multiple transformations.</p>
-
-<h3 id="map">Map</h3>
-
-<p>The Map transformation applies a user-defined <code>MapFunction</code> on each element of a DataSet.
-It implements a one-to-one mapping, that is, exactly one element must be returned by
-the function.</p>
-
-<p>The following code transforms a <code>DataSet</code> of Integer pairs into a <code>DataSet</code> of Integers:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// MapFunction that adds two integer values</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">IntAdder</span> <span class="kd">extends</span> <span class="n">MapFunction</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>,</span> <span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">map</span><span class="o">(</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="n">in</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">return</span> <span class="n">in</span><span class="o">.</span><span class="na">f0</span> <span class="o">+</span> <span class="n">in</span><span class="o">.</span><span class="na">f1</span><span class="o">;</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="n">intPairs</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">intSums</span> <span class="o">=</span> <span class="n">intPairs</span><span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="nf">IntAdder</span><span class="o">());</span>
-</code></pre></div>
-<h3 id="flatmap">FlatMap</h3>
-
-<p>The FlatMap transformation applies a user-defined <code>FlatMapFunction</code> on each element of a <code>DataSet</code>.
-This variant of a map function can return arbitrary many result elements (including none) for each input element.</p>
-
-<p>The following code transforms a <code>DataSet</code> of text lines into a <code>DataSet</code> of words:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// FlatMapFunction that tokenizes a String by whitespace characters and emits all String tokens.</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">Tokenizer</span> <span class="kd">extends</span> <span class="n">FlatMapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">flatMap</span><span class="o">(</span><span class="n">String</span> <span class="n">value</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">for</span> <span class="o">(</span><span class="n">String</span> <span class="n">token</span> <span class="o">:</span> <span class="n">value</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">"\\W"</span><span class="o">))</span> <span class="o">{</span>
- <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">token</span><span class="o">);</span>
- <span class="o">}</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">textLines</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">words</span> <span class="o">=</span> <span class="n">textLines</span><span class="o">.</span><span class="na">flatMap</span><span class="o">(</span><span class="k">new</span> <span class="nf">Tokenizer</span><span class="o">());</span>
-</code></pre></div>
-<h3 id="filter">Filter</h3>
-
-<p>The Filter transformation applies a user-defined <code>FilterFunction</code> on each element of a <code>DataSet</code> and retains only those elements for which the function returns <code>true</code>.</p>
-
-<p>The following code removes all Integers smaller than zero from a <code>DataSet</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// FilterFunction that filters out all Integers smaller than zero.</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">NaturalNumberFilter</span> <span class="kd">extends</span> <span class="n">FilterFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">boolean</span> <span class="nf">filter</span><span class="o">(</span><span class="n">Integer</span> <span class="n">number</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">return</span> <span class="n">number</span> <span class="o">>=</span> <span class="mi">0</span><span class="o">;</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">intNumbers</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">naturalNumbers</span> <span class="o">=</span> <span class="n">intNumbers</span><span class="o">.</span><span class="na">filter</span><span class="o">(</span><span class="k">new</span> <span class="nf">NaturalNumberFilter</span><span class="o">());</span>
-</code></pre></div>
-<h3 id="project-(tuple-datasets-only)">Project (Tuple DataSets only)</h3>
-
-<p>The Project transformation removes or moves <code>Tuple</code> fields of a <code>Tuple</code> <code>DataSet</code>.
-The <code>project(int...)</code> method selects <code>Tuple</code> fields that should be retained by their index and defines their order in the output <code>Tuple</code>.
-The <code>types(Class<?> ...)</code>method must give the types of the output <code>Tuple</code> fields.</p>
-
-<p>Projections do not require the definition of a user function.</p>
-
-<p>The following code shows different ways to apply a Project transformation on a <code>DataSet</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">in</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="c1">// converts Tuple3<Integer, Double, String> into Tuple2<String, Integer></span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="n">out</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">project</span><span class="o">(</span><span class="mi">2</span><span class="o">,</span><span class="mi">0</span><span class="o">).</span><span class="na">types</span><span class="o">(</span><span class="n">String</span><span class="o">.</span><span class="na">class</span><span class="o">,</span> <span class="n">Integer</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
-</code></pre></div>
-<h3 id="transformations-on-grouped-dataset">Transformations on grouped DataSet</h3>
-
-<p>The reduce operations can operate on grouped data sets. Specifying the key to
-be used for grouping can be done in two ways:</p>
-
-<ul>
-<li>a <code>KeySelector</code> function or</li>
-<li>one or more field position keys (<code>Tuple</code> <code>DataSet</code> only).</li>
-</ul>
-
-<p>Please look at the reduce examples to see how the grouping keys are specified.</p>
-
-<h3 id="reduce-on-grouped-dataset">Reduce on grouped DataSet</h3>
-
-<p>A Reduce transformation that is applied on a grouped <code>DataSet</code> reduces each group to a single element using a user-defined <code>ReduceFunction</code>.
-For each group of input elements, a <code>ReduceFunction</code> successively combines pairs of elements into one element until only a single element for each group remains.</p>
-
-<h4 id="reduce-on-dataset-grouped-by-keyselector-function">Reduce on DataSet grouped by KeySelector Function</h4>
-
-<p>A <code>KeySelector</code> function extracts a key value from each element of a <code>DataSet</code>. The extracted key value is used to group the <code>DataSet</code>.
-The following code shows how to group a POJO <code>DataSet</code> using a <code>KeySelector</code> function and to reduce it with a <code>ReduceFunction</code>.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// some ordinary POJO</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">WC</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="n">word</span><span class="o">;</span>
- <span class="kd">public</span> <span class="kt">int</span> <span class="n">count</span><span class="o">;</span>
- <span class="c1">// [...]</span>
-<span class="o">}</span>
-
-<span class="c1">// ReduceFunction that sums Integer attributes of a POJO</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">WordCounter</span> <span class="kd">extends</span> <span class="n">ReduceFunction</span><span class="o"><</span><span class="n">WC</span><span class="o">></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="n">WC</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">WC</span> <span class="n">in1</span><span class="o">,</span> <span class="n">WC</span> <span class="n">in2</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">return</span> <span class="k">new</span> <span class="nf">WC</span><span class="o">(</span><span class="n">in1</span><span class="o">.</span><span class="na">word</span><span class="o">,</span> <span class="n">in1</span><span class="o">.</span><span class="na">count</span> <span class="o">+</span> <span class="n">in2</span><span class="o">.</span><span class="na">count</span><span class="o">);</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">WC</span><span class="o">></span> <span class="n">words</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">WC</span><span class="o">></span> <span class="n">wordCounts</span> <span class="o">=</span> <span class="n">words</span>
- <span class="c1">// DataSet grouping with inline-defined KeySelector function</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span>
- <span class="k">new</span> <span class="n">KeySelector</span><span class="o"><</span><span class="n">WC</span><span class="o">,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="nf">getKey</span><span class="o">(</span><span class="n">WC</span> <span class="n">wc</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">wc</span><span class="o">.</span><span class="na">word</span><span class="o">;</span> <span class="o">}</span>
- <span class="o">})</span>
- <span class="c1">// apply ReduceFunction on grouped DataSet</span>
- <span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="k">new</span> <span class="nf">WordCounter</span><span class="o">());</span>
-</code></pre></div>
-<h4 id="reduce-on-dataset-grouped-by-field-position-keys-(tuple-datasets-only)">Reduce on DataSet grouped by Field Position Keys (Tuple DataSets only)</h4>
-
-<p>Field position keys specify one or more fields of a <code>Tuple</code> <code>DataSet</code> that are used as grouping keys.
-The following code shows how to use field position keys and apply a <code>ReduceFunction</code>.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">tuples</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">reducedTuples</span> <span class="o">=</span>
- <span class="n">tuples</span>
- <span class="c1">// group DataSet on first and second field of Tuple</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span><span class="mi">1</span><span class="o">)</span>
- <span class="c1">// apply ReduceFunction on grouped DataSet</span>
- <span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="k">new</span> <span class="nf">MyTupleReducer</span><span class="o">());</span>
-</code></pre></div>
-<h3 id="groupreduce-on-grouped-dataset">GroupReduce on grouped DataSet</h3>
-
-<p>A GroupReduce transformation that is applied on a grouped <code>DataSet</code> calls a user-defined <code>GroupReduceFunction</code> for each group. The difference
-between this and <code>Reduce</code> is that the user defined function gets the whole group at once.
-The function is invoked with an iterator over all elements of a group and can return an arbitrary number of result elements using the collector.</p>
-
-<h4 id="groupreduce-on-dataset-grouped-by-field-position-keys-(tuple-datasets-only)">GroupReduce on DataSet grouped by Field Position Keys (Tuple DataSets only)</h4>
-
-<p>The following code shows how duplicate strings can be removed from a <code>DataSet</code> grouped by Integer.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">class</span> <span class="nc">DistinctReduce</span>
- <span class="kd">extends</span> <span class="n">GroupReduceFunction</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="o">{</span>
- <span class="c1">// Set to hold all unique strings of a group</span>
- <span class="n">Set</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">uniqStrings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">HashSet</span><span class="o"><</span><span class="n">String</span><span class="o">>();</span>
-
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Iterator</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">in</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
- <span class="c1">// clear set</span>
- <span class="n">uniqStrings</span><span class="o">.</span><span class="na">clear</span><span class="o">();</span>
- <span class="c1">// there is at least one element in the iterator</span>
- <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">></span> <span class="n">first</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">();</span>
- <span class="n">Integer</span> <span class="n">key</span> <span class="o">=</span> <span class="n">first</span><span class="o">.</span><span class="na">f0</span><span class="o">;</span>
- <span class="n">uniqStrings</span><span class="o">.</span><span class="na">add</span><span class="o">(</span><span class="n">first</span><span class="o">.</span><span class="na">f1</span><span class="o">);</span>
- <span class="c1">// add all strings of the group to the set</span>
- <span class="k">while</span><span class="o">(</span><span class="n">in</span><span class="o">.</span><span class="na">hasNext</span><span class="o">())</span> <span class="o">{</span>
- <span class="n">uniqStrings</span><span class="o">.</span><span class="na">add</span><span class="o">(</span><span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">().</span><span class="na">f1</span><span class="o">);</span>
- <span class="o">}</span>
- <span class="c1">// emit all unique strings</span>
- <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">></span> <span class="n">t</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>(</span><span class="n">key</span><span class="o">,</span> <span class="s">""</span><span class="o">);</span>
- <span class="k">for</span><span class="o">(</span><span class="n">String</span> <span class="n">s</span> <span class="o">:</span> <span class="n">uniqStrings</span><span class="o">)</span> <span class="o">{</span>
- <span class="n">t</span><span class="o">.</span><span class="na">f1</span> <span class="o">=</span> <span class="n">s</span><span class="o">;</span>
- <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="n">t</span><span class="o">);</span>
- <span class="o">}</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">output</span> <span class="o">=</span>
- <span class="n">input</span>
- <span class="c1">// group DataSet by the first tuple field</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
- <span class="c1">// apply GroupReduceFunction on each group and</span>
- <span class="c1">// remove elements with duplicate strings.</span>
- <span class="o">.</span><span class="na">reduceGroup</span><span class="o">(</span><span class="k">new</span> <span class="nf">DistinctReduce</span><span class="o">());</span>
-</code></pre></div>
-<p><strong>Note:</strong> Flink internally works a lot with mutable objects. Collecting objects like in the above example only works because Strings are immutable in Java!</p>
-
-<h4 id="groupreduce-on-dataset-grouped-by-keyselector-function">GroupReduce on DataSet grouped by KeySelector Function</h4>
-
-<p>Works analogous to <code>KeySelector</code> functions in Reduce transformations.</p>
-
-<h4 id="groupreduce-on-sorted-groups-(tuple-datasets-only)">GroupReduce on sorted groups (Tuple DataSets only)</h4>
-
-<p>A <code>GroupReduceFunction</code> accesses the elements of a group using an iterator. Optionally, the iterator can hand out the elements of a group in a specified order. In many cases this can help to reduce the complexity of a user-defined <code>GroupReduceFunction</code> and improve its efficiency.
-Right now, this feature is only available for <code>Tuple</code> <code>DataSet</code>.</p>
-
-<p>The following code shows another example how to remove duplicate Strings in a <code>DataSet</code> grouped by an Integer and sorted by String.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// GroupReduceFunction that removes consecutive identical elements</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">DistinctReduce</span>
- <span class="kd">extends</span> <span class="n">GroupReduceFunction</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Iterator</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">in</span><span class="o">,</span> <span class="n">Collector</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
- <span class="c1">// there is at least one element in the iterator</span>
- <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">></span> <span class="n">first</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">();</span>
- <span class="n">Integer</span> <span class="n">key</span> <span class="o">=</span> <span class="n">first</span><span class="o">.</span><span class="na">f0</span><span class="o">;</span>
- <span class="n">String</span> <span class="n">comp</span> <span class="o">=</span> <span class="n">first</span><span class="o">.</span><span class="na">f1</span><span class="o">;</span>
- <span class="c1">// for each element in group</span>
- <span class="k">while</span><span class="o">(</span><span class="n">in</span><span class="o">.</span><span class="na">hasNext</span><span class="o">())</span> <span class="o">{</span>
- <span class="n">String</span> <span class="n">next</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">().</span><span class="na">f1</span><span class="o">;</span>
- <span class="c1">// check if strings are different</span>
- <span class="k">if</span><span class="o">(!</span><span class="n">next</span><span class="o">.</span><span class="na">equals</span><span class="o">(</span><span class="n">comp</span><span class="o">))</span> <span class="o">{</span>
- <span class="c1">// emit a new element</span>
- <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="k">new</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>(</span><span class="n">key</span><span class="o">,</span> <span class="n">comp</span><span class="o">));</span>
- <span class="c1">// update compare string</span>
- <span class="n">comp</span> <span class="o">=</span> <span class="n">next</span><span class="o">;</span>
- <span class="o">}</span>
- <span class="o">}</span>
- <span class="c1">// emit last element</span>
- <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="k">new</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>(</span><span class="n">key</span><span class="o">,</span> <span class="n">comp</span><span class="o">));</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Double</span><span class="o">></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span>
- <span class="c1">// group DataSet by the first tuple field</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
- <span class="c1">// sort groups on second tuple field</span>
- <span class="o">.</span><span class="na">sortGroup</span><span class="o">(</span><span class="mi">1</span><span class="o">,</span> <span class="n">Order</span><span class="o">.</span><span class="na">ASCENDING</span><span class="o">)</span>
- <span class="c1">// // apply GroupReduceFunction on DataSet with sorted groups</span>
- <span class="o">.</span><span class="na">reduceGroup</span><span class="o">(</span><span class="k">new</span> <span class="nf">DistinctReduce</span><span class="o">());</span>
-</code></pre></div>
-<p><strong>Note:</strong> A GroupSort often comes for free if the grouping is established using a sort-based execution strategy of an operator before the reduce operation.</p>
-
-<h4 id="combinable-groupreducefunctions">Combinable GroupReduceFunctions</h4>
-
-<p>In contrast to a <code>ReduceFunction</code>, a <code>GroupReduceFunction</code> is not implicitly combinable. In order to make a <code>GroupReduceFunction</code> combinable, you need to implement (override) the <code>combine()</code> method and annotate the <code>GroupReduceFunction</code> with the <code>@Combinable</code> annotation as shown here:</p>
-
-<p>The following code shows how to compute multiple sums using a combinable <code>GroupReduceFunction</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// Combinable GroupReduceFunction that computes two sums.</span>
-<span class="nd">@Combinable</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyCombinableGroupReducer</span>
- <span class="kd">extends</span> <span class="n">GroupReduceFunction</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>,</span>
- <span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Iterator</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">in</span><span class="o">,</span>
- <span class="n">Collector</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">out</span><span class="o">)</span> <span class="o">{</span>
- <span class="c1">// one element is always present in iterator</span>
- <span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">></span> <span class="n">curr</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">();</span>
- <span class="n">String</span> <span class="n">key</span> <span class="o">=</span> <span class="n">curr</span><span class="o">.</span><span class="na">f0</span><span class="o">;</span>
- <span class="kt">int</span> <span class="n">intSum</span> <span class="o">=</span> <span class="n">curr</span><span class="o">.</span><span class="na">f1</span><span class="o">;</span>
- <span class="kt">double</span> <span class="n">doubleSum</span> <span class="o">=</span> <span class="n">curr</span><span class="o">.</span><span class="na">f2</span><span class="o">;</span>
- <span class="c1">// sum up all ints and doubles</span>
- <span class="k">while</span><span class="o">(</span><span class="n">in</span><span class="o">.</span><span class="na">hasNext</span><span class="o">())</span> <span class="o">{</span>
- <span class="n">curr</span> <span class="o">=</span> <span class="n">in</span><span class="o">.</span><span class="na">next</span><span class="o">();</span>
- <span class="n">intSum</span> <span class="o">+=</span> <span class="n">curr</span><span class="o">.</span><span class="na">f1</span><span class="o">;</span>
- <span class="n">doubleSum</span> <span class="o">+=</span> <span class="n">curr</span><span class="o">.</span><span class="na">f2</span><span class="o">;</span>
- <span class="o">}</span>
- <span class="c1">// emit a tuple with both sums</span>
- <span class="n">out</span><span class="o">.</span><span class="na">collect</span><span class="o">(</span><span class="k">new</span> <span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>(</span><span class="n">key</span><span class="o">,</span> <span class="n">intSum</span><span class="o">,</span> <span class="n">doubleSum</span><span class="o">));</span>
- <span class="o">}</span>
-
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">combine</span><span class="o">(</span><span class="n">Iterator</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">in</span><span class="o">,</span>
- <span class="n">Collector</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">out</span><span class="o">))</span> <span class="o">{</span>
- <span class="c1">// in some cases combine() calls can simply be forwarded to reduce().</span>
- <span class="k">this</span><span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="n">in</span><span class="o">,</span> <span class="n">out</span><span class="o">);</span>
- <span class="o">}</span>
-<span class="o">}</span>
-</code></pre></div>
-<h3 id="aggregate-on-grouped-tuple-dataset">Aggregate on grouped Tuple DataSet</h3>
-
-<p>There are some common aggregation operations that are frequently used. The Aggregate transformation provides the following build-in aggregation functions:</p>
-
-<ul>
-<li>Sum,</li>
-<li>Min, and</li>
-<li>Max.</li>
-</ul>
-
-<p>The Aggregate transformation can only be applied on a <code>Tuple</code> <code>DataSet</code> and supports only field positions keys for grouping.</p>
-
-<p>The following code shows how to apply an Aggregation transformation on a <code>DataSet</code> grouped by field position keys:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span>
- <span class="c1">// group DataSet on second field</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="mi">1</span><span class="o">)</span>
- <span class="c1">// compute sum of the first field</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="n">SUM</span><span class="o">,</span> <span class="mi">0</span><span class="o">)</span>
- <span class="c1">// compute minimum of the third field</span>
- <span class="o">.</span><span class="na">and</span><span class="o">(</span><span class="n">MIN</span><span class="o">,</span> <span class="mi">2</span><span class="o">);</span>
-</code></pre></div>
-<p>To apply multiple aggregations on a DataSet it is necessary to use the <code>.and()</code> function after the first aggregate, that means <code>.aggregate(SUM, 0).and(MIN, 2)</code> produces the sum of field 0 and the minimum of field 2 of the original DataSet.
-In contrast to that <code>.aggregate(SUM, 0).aggregate(MIN, 2)</code> will apply an aggregation on an aggregation. In the given example it would produce the minimum of field 2 after calculating the sum of field 0 grouped by field 1.</p>
-
-<p><strong>Note:</strong> The set of aggregation functions will be extended in the future.</p>
-
-<h3 id="reduce-on-full-dataset">Reduce on full DataSet</h3>
-
-<p>The Reduce transformation applies a user-defined <code>ReduceFunction</code> to all elements of a <code>DataSet</code>.
-The <code>ReduceFunction</code> subsequently combines pairs of elements into one element until only a single element remains.</p>
-
-<p>The following code shows how to sum all elements of an Integer <code>DataSet</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// ReduceFunction that sums Integers</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">IntSummer</span> <span class="kd">extends</span> <span class="n">ReduceFunction</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Integer</span> <span class="n">num1</span><span class="o">,</span> <span class="n">Integer</span> <span class="n">num2</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">return</span> <span class="n">num1</span> <span class="o">+</span> <span class="n">num2</span><span class="o">;</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">intNumbers</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">sum</span> <span class="o">=</span> <span class="n">intNumbers</span><span class="o">.</span><span class="na">reduce</span><span class="o">(</span><span class="k">new</span> <span class="nf">IntSummer</span><span class="o">());</span>
-</code></pre></div>
-<p>Reducing a full <code>DataSet</code> using the Reduce transformation implies that the final Reduce operation cannot be done in parallel. However, a <code>ReduceFunction</code> is automatically combinable such that a Reduce transformation does not limit scalability for most use cases.</p>
-
-<h3 id="groupreduce-on-full-dataset">GroupReduce on full DataSet</h3>
-
-<p>The GroupReduce transformation applies a user-defined <code>GroupReduceFunction</code> on all elements of a <code>DataSet</code>.
-A <code>GroupReduceFunction</code> can iterate over all elements of <code>DataSet</code> and return an arbitrary number of result elements.</p>
-
-<p>The following example shows how to apply a GroupReduce transformation on a full <code>DataSet</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="c1">// apply a (preferably combinable) GroupReduceFunction to a DataSet</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Double</span><span class="o">></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span><span class="o">.</span><span class="na">reduceGroup</span><span class="o">(</span><span class="k">new</span> <span class="nf">MyGroupReducer</span><span class="o">());</span>
-</code></pre></div>
-<p><strong>Note:</strong> A GroupReduce transformation on a full <code>DataSet</code> cannot be done in parallel if the <code>GroupReduceFunction</code> is not combinable. Therefore, this can be a very compute intensive operation. See the paragraph on "Combineable <code>GroupReduceFunction</code>s" above to learn how to implement a combinable <code>GroupReduceFunction</code>.</p>
-
-<h3 id="aggregate-on-full-tuple-dataset">Aggregate on full Tuple DataSet</h3>
-
-<p>There are some common aggregation operations that are frequently used. The Aggregate transformation provides the following build-in aggregation functions:</p>
-
-<ul>
-<li>Sum,</li>
-<li>Min, and</li>
-<li>Max.</li>
-</ul>
-
-<p>The Aggregate transformation can only be applied on a <code>Tuple</code> <code>DataSet</code>.</p>
-
-<p>The following code shows how to apply an Aggregation transformation on a full <code>DataSet</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">input</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">output</span> <span class="o">=</span> <span class="n">input</span>
- <span class="c1">// compute sum of the first field</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="n">SUM</span><span class="o">,</span> <span class="mi">0</span><span class="o">)</span>
- <span class="c1">// compute minimum of the second field</span>
- <span class="o">.</span><span class="na">and</span><span class="o">(</span><span class="n">MIN</span><span class="o">,</span> <span class="mi">1</span><span class="o">);</span>
-</code></pre></div>
-<p><strong>Note:</strong> Extending the set of supported aggregation functions is on our roadmap.</p>
-
-<h3 id="join">Join</h3>
-
-<p>The Join transformation joins two <code>DataSet</code>s into one <code>DataSet</code>. The elements of both <code>DataSet</code>s are joined on one or more keys which can be specified using</p>
-
-<ul>
-<li>a <code>KeySelector</code> function or</li>
-<li>one or more field position keys (<code>Tuple</code> <code>DataSet</code> only).</li>
-</ul>
-
-<p>There are a few different ways to perform a Join transformation which are shown in the following.</p>
-
-<h4 id="default-join-(join-into-tuple2)">Default Join (Join into Tuple2)</h4>
-
-<p>The default Join transformation produces a new <code>Tuple</code><code>DataSet</code> with two fields. Each tuple holds a joined element of the first input <code>DataSet</code> in the first tuple field and a matching element of the second input <code>DataSet</code> in the second field.</p>
-
-<p>The following code shows a default Join transformation using field position keys:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input1</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Double</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>></span> <span class="n">input2</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="c1">// result dataset is typed as Tuple2</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">Double</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>>></span>
- <span class="n">result</span> <span class="o">=</span>
- <span class="n">input1</span><span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">input2</span><span class="o">)</span>
- <span class="c1">// key definition on first DataSet using a field position key</span>
- <span class="o">.</span><span class="na">where</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
- <span class="c1">// key definition of second DataSet using a field position key</span>
- <span class="o">.</span><span class="na">equalTo</span><span class="o">(</span><span class="mi">1</span><span class="o">);</span>
-</code></pre></div>
-<h4 id="join-with-joinfunction">Join with JoinFunction</h4>
-
-<p>A Join transformation can also call a user-defined <code>JoinFunction</code> to process joining tuples.
-A <code>JoinFunction</code> receives one element of the first input <code>DataSet</code> and one element of the second input <code>DataSet</code> and returns exactly one element.</p>
-
-<p>The following code performs a join of <code>DataSet</code> with custom java objects and a <code>Tuple</code> <code>DataSet</code> using <code>KeySelector</code> functions and shows how to call a user-defined <code>JoinFunction</code>:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// some POJO</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">Rating</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="n">name</span><span class="o">;</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="n">category</span><span class="o">;</span>
- <span class="kd">public</span> <span class="kt">int</span> <span class="n">points</span><span class="o">;</span>
-<span class="o">}</span>
-
-<span class="c1">// Join function that joins a custom POJO with a Tuple</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">PointWeighter</span>
- <span class="kd">extends</span> <span class="n">JoinFunction</span><span class="o"><</span><span class="n">Rating</span><span class="o">,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="o">{</span>
-
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">></span> <span class="nf">join</span><span class="o">(</span><span class="n">Rating</span> <span class="n">rating</span><span class="o">,</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">></span> <span class="n">weight</span><span class="o">)</span> <span class="o">{</span>
- <span class="c1">// multiply the points and rating and construct a new output tuple</span>
- <span class="k">return</span> <span class="k">new</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>(</span><span class="n">rating</span><span class="o">.</span><span class="na">name</span><span class="o">,</span> <span class="n">rating</span><span class="o">.</span><span class="na">points</span> <span class="o">*</span> <span class="n">weight</span><span class="o">.</span><span class="na">f1</span><span class="o">);</span>
- <span class="o">}</span>
-<span class="o">}</span>
-
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Rating</span><span class="o">></span> <span class="n">ratings</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">weights</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span>
- <span class="n">weightedRatings</span> <span class="o">=</span>
- <span class="n">ratings</span><span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">weights</span><span class="o">)</span>
- <span class="c1">// key definition of first DataSet using a KeySelector function</span>
- <span class="o">.</span><span class="na">where</span><span class="o">(</span><span class="k">new</span> <span class="n">KeySelection</span><span class="o"><</span><span class="n">Rating</span><span class="o">,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="nf">getKey</span><span class="o">(</span><span class="n">Rating</span> <span class="n">r</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">r</span><span class="o">.</span><span class="na">category</span><span class="o">;</span> <span class="o">}</span>
- <span class="o">})</span>
- <span class="c1">// key definition of second DataSet using a KeySelector function</span>
- <span class="o">.</span><span class="na">equalTo</span><span class="o">(</span><span class="k">new</span> <span class="n">KeySelection</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">>,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">String</span> <span class="nf">getKey</span><span class="o">(</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">></span> <span class="n">t</span><span class="o">)</span> <span class="o">{</span> <span class="k">return</span> <span class="n">t</span><span class="o">.</span><span class="na">f0</span><span class="o">;</span> <span class="o">}</span>
- <span class="o">})</span>
- <span class="c1">// applying the JoinFunction on joining pairs</span>
- <span class="o">.</span><span class="na">with</span><span class="o">(</span><span class="k">new</span> <span class="nf">PointWeighter</span><span class="o">());</span>
-</code></pre></div>
-<h4 id="join-with-projection">Join with Projection</h4>
-
-<p>A Join transformation can construct result tuples using a projection as shown here:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple3</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Byte</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input1</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Double</span><span class="o">>></span> <span class="n">input2</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple4</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Double</span><span class="o">,</span> <span class="n">Byte</span><span class="o">></span>
- <span class="n">result</span> <span class="o">=</span>
- <span class="n">input1</span><span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">input2</span><span class="o">)</span>
- <span class="c1">// key definition on first DataSet using a field position key</span>
- <span class="o">.</span><span class="na">where</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
- <span class="c1">// key definition of second DataSet using a field position key</span>
- <span class="o">.</span><span class="na">equalTo</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span>
- <span class="c1">// select and reorder fields of matching tuples</span>
- <span class="o">.</span><span class="na">projectFirst</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span><span class="mi">2</span><span class="o">).</span><span class="na">projectSecond</span><span class="o">(</span><span class="mi">1</span><span class="o">).</span><span class="na">projectFirst</span><span class="o">(</span><span class="mi">1</span><span class="o">)</span>
- <span class="o">.</span><span class="na">types</span><span class="o">(</span><span class="n">Integer</span><span class="o">.</span><span class="na">class</span><span class="o">,</span> <span class="n">String</span><span class="o">.</span><span class="na">class</span><span class="o">,</span> <span class="n">Double</span><span class="o">.</span><span class="na">class</span><span class="o">,</span> <span class="n">Byte</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
-</code></pre></div>
-<p><code>projectFirst(int...)</code> and <code>projectSecond(int...)</code> select the fields of the first and second joined input that should be assembled into an output <code>Tuple</code>. The order of indexes defines the order of fields in the output tuple.
-The join projection works also for non-<code>Tuple</code> <code>DataSet</code>s. In this case, <code>projectFirst()</code> or <code>projectSecond()</code> must be called without arguments to add a joined element to the output <code>Tuple</code>.</p>
-
-<h4 id="join-with-dataset-size-hint">Join with DataSet Size Hint</h4>
-
-<p>In order to guide the optimizer to pick the right execution strategy, you can hint the size of a <code>DataSet</code> to join as shown here:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input1</span> <span class="o">=</span> <span class="c1">// [...]</span>
-<span class="n">DataSet</span><span class="o"><</span><span class="n">Tuple2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">String</span><span class="o">>></span> <span class="n">input2</span> <span class="o">=</span> <span class="c1">// [...]</span>
-
[... 172 lines stripped ...]