You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2014/10/08 12:23:45 UTC
[5/6] git commit: [FLINK-1107] Hadoop Compatibility Layer documented
[FLINK-1107] Hadoop Compatibility Layer documented
This closes #131
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/3ebdebc0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/3ebdebc0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/3ebdebc0
Branch: refs/heads/master
Commit: 3ebdebc0ad07ce0693a717297aa1c50bb163a868
Parents: aa6d1cf
Author: twalthr <in...@twalthr.com>
Authored: Mon Sep 29 14:01:17 2014 +0200
Committer: Fabian Hueske <fh...@apache.org>
Committed: Wed Oct 8 12:22:27 2014 +0200
----------------------------------------------------------------------
docs/_includes/sidenav.html | 1 +
docs/hadoop_compatibility.md | 74 +++++++++++++++++++++++++++++++++++++--
docs/spargel_guide.md | 6 ++--
3 files changed, 76 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3ebdebc0/docs/_includes/sidenav.html
----------------------------------------------------------------------
diff --git a/docs/_includes/sidenav.html b/docs/_includes/sidenav.html
index 7c22633..5797ecc 100644
--- a/docs/_includes/sidenav.html
+++ b/docs/_includes/sidenav.html
@@ -27,6 +27,7 @@
<li><a href="streaming_guide.html">Streaming Guide</a></li>
<li><a href="iterations.html">Iterations</a></li>
<li><a href="spargel_guide.html">Spargel Graph API</a></li>
+ <li><a href="hadoop_compatibility.html">Hadoop I/O Compatibility</a></li>
</ul>
</li>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3ebdebc0/docs/hadoop_compatibility.md
----------------------------------------------------------------------
diff --git a/docs/hadoop_compatibility.md b/docs/hadoop_compatibility.md
index 06c0dfa..7b2c217 100644
--- a/docs/hadoop_compatibility.md
+++ b/docs/hadoop_compatibility.md
@@ -1,5 +1,75 @@
---
-title: "Hadoop Compatability"
+title: "Hadoop I/O Compatibility"
---
-To be written.
\ No newline at end of file
+Flink not only supports types that implement Apache Hadoop's `Writable` interface by default, but also provides a compatibility
+layer that allows for using any class extending `org.apache.hadoop.mapred(uce).InputFormat` as a Flink `InputFormat` as well as any
+class extending `org.apache.hadoop.mapred(uce).OutputFormat` as a Flink `OutputFormat`.
+
+Thus, Flink can handle Hadoop-related formats from the common `TextInputFormat` up to third-party components such as Hive through HCatalog's `HCatInputFormat`. Flink supports both the formats which use the old `org.apache.hadoop.mapred` API as well as the new `org.apache.hadoop.mapreduce` API.
+
+This document explains how to configure your Maven project correctly and shows an example.
+
+### Project Configuration
+
+The Hadoop Compatibility Layer is part of the *addons* Maven project. All relevant classes are located in the `org.apache.flink.hadoopcompatibility` package. The package includes separate packages and classes for the Hadoop `mapred` and `mapreduce` API.
+
+Add the following dependency to your `pom.xml` to use the Hadoop Compatibility Layer.
+
+~~~xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-hadoop-compatibility</artifactId>
+ <version>{{site.FLINK_VERSION_STABLE}}</version>
+</dependency>
+~~~
+
+### Examples
+
+The following example shows how to read a file from Hadoop's `TextInputFormat`, count the words with Flink and output the result with Hadoop's `TextOutputFormat`.
+
+Flink's `HadoopInputFormat` is initiated in line 5 and acts as a wrapper for the Hadoop `TextInputFormat`. An instance of `TextInputFormat` and the corresponding return types for key and value together with a Hadoop `Job` definition must be passed to the constructor of the `HadoopInputFormat`.
+
+Take care that you choose the right classes since `HadoopInputFormat` and `HadoopOutputFormat` exist in both packages for Hadoop's `mapred` and `mapreduce` API. For example, the `HadoopInputFormat` for the `mapred` API takes a Hadoop `JobConf` instance as parameter instead of `Job` instance.
+
+Flink's `HadoopOutputFormat` is initiated in a similar way with an instance of Hadoop's `TextOutputFormat` and the previously used Hadoop `Job` definition.
+
+Additional Hadoop properties can be set by calling the formats `getConfiguration()` (for `mapreduce` API) or `getJobConf()` (for `mapred` API) method.
+
+~~~java
+ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+// Set up the Hadoop Input Format
+Job job = Job.getInstance();
+HadoopInputFormat<LongWritable, Text> hadoopInputFormat =
+ new HadoopInputFormat<LongWritable, Text>(
+ new TextInputFormat(), LongWritable.class, Text.class, job
+ );
+TextInputFormat.addInputPath(job, new Path(inputPath));
+
+// Create a Flink job with it
+DataSet<Tuple2<LongWritable, Text>> text = env.createInput(hadoopInputFormat);
+
+// Tokenize the line and convert from Writable "Text" to String for better handling
+DataSet<Tuple2<String, Integer>> words = text.flatMap(new Tokenizer());
+
+// Sum up the words
+DataSet<Tuple2<String, Integer>> result = words.groupBy(0).aggregate(Aggregations.SUM, 1);
+
+// Convert String back to Writable "Text" for use with Hadoop Output Format
+DataSet<Tuple2<Text, IntWritable>> hadoopResult = result.map(new HadoopDatatypeMapper());
+
+// Set up Hadoop Output Format
+HadoopOutputFormat<Text, IntWritable> hadoopOutputFormat =
+ new HadoopOutputFormat<Text, IntWritable>(
+ new TextOutputFormat<Text, IntWritable>(), job
+ );
+hadoopOutputFormat.getConfiguration().set("mapreduce.output.textoutputformat.separator", " ");
+TextOutputFormat.setOutputPath(job, new Path(outputPath));
+
+// Output & Execute
+hadoopResult.output(hadoopOutputFormat);
+env.execute("Word Count");
+~~~
+
+A full running example can be found in `org.apache.flink.hadoopcompatibility.mapred(uce).example.WordCount`.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3ebdebc0/docs/spargel_guide.md
----------------------------------------------------------------------
diff --git a/docs/spargel_guide.md b/docs/spargel_guide.md
index 6fe2585..c5b2ac4 100644
--- a/docs/spargel_guide.md
+++ b/docs/spargel_guide.md
@@ -17,14 +17,14 @@ This vertex-centric view makes it easy to express a large class of graph problem
Spargel API
-----------
-The Spargel API is part of the *addons* Maven project. All relevant classes are located in the *org.apache.flinkspargel.java* package.
+The Spargel API is part of the *addons* Maven project. All relevant classes are located in the *org.apache.flink.spargel.java* package.
Add the following dependency to your `pom.xml` to use the Spargel.
~~~xml
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>spargel</artifactId>
+ <artifactId>flink-spargel</artifactId>
<version>{{site.FLINK_VERSION_STABLE}}</version>
</dependency>
~~~
@@ -110,4 +110,4 @@ The computation **terminates** after a specified *maximum number of supersteps*
<p class="text-center">
<img alt="Spargel Example" width="75%" src="img/spargel_example.png" />
-</p>
\ No newline at end of file
+</p>