stratosphere/stratosphere — pull request #312 by rmetzger (2013-12-02 18:58:59+00:00)
Title: Experimental YARN support for Stratosphere

This pull request contains the code required to submit Stratosphere to a Hadoop (Yarn|2.2|NextGen) cluster.

How to:
Build the Stratosphere-yarn-uberjar:
```
mvn package -Dhadoop.profile=2 -DskipTests
```
Run it
```
cd stratosphere-dist/target
java -jar stratosphere-dist-0.4-SNAPSHOT-yarn-uberjar.jar
Missing required option: [-n Number of Yarn container to allocate (=Number of TaskTrackers)]
Usage:
Required
-n,--container Number of Yarn container to allocate (=Number of TaskTrackers)
Optional
-c,--conf Path to Stratosphere configuration file
-j,--jar Path to Stratosphere jar file
-jm,--jobManagerMemory Memory for JobManager Container [in MB]
-tm,--taskManagerMemory Memory per TaskManager Container [in MB]
-tmc,--taskManagerCores Virtual CPU cores per TaskManager
-v,--verbose Verbose debug mode

```

The jar will automatically create a local `stratosphere-config.yml` in the local directory and use this.
The only requirement for this to run is that the `HADOOP_HOME` environment variable is set. My client is able to read the hadoop configuration from there and talk to the yarn resource manager.
Yarn must be set up with a HDFS, since I'm using HDFS to distribute the Stratosphere-jar and the modified configuration file (the config for the TaskManagers must me changed to have the container of the JobManager contained)
The only required argument is the number of TaskManagers (`-n`).
The JobManager runs as a separate thread in the ApplicationMaster. So the number of containers in yarn is always `n+1`.

Submit a job to the yarn stratosphere: You can just use the regular mechanisms to submit jobs. I also extended the pact-client.sh to support the `RemoteExecutor` written by @aljoscha.

```
java -cp stratosphere-dist-0.4-SNAPSHOT-yarn-uberjar.jar eu.stratosphere.pact.client.CliFrontend remote cloud-13:6123 /home/rmetzger/stratosphere-tutorial-reference-0.1-SNAPSHOT.jar \
eu.stratosphere.tutorial.task4.WeightVectorPlan \
"hdfs:///user/robert/bigdataclass-wikipedia hdfs:///user/robert/bigdataclass-wikipedia-result $DOP"
```
The pact-client.sh with remote submission is undocumented and only hacked into the client. It is used like this:
```
Usage: [host:port] [jar] [class] [args]
```

What's missing
* Extensive testing (I tested it locally, and on the Dima cluster with 4 nodes (and the tf-idf job (10 GB))
* Proper tear-down (currently, you have to use yarn to kill the application)
* More robustness towards failure: The whole thing breaks if one container is killed or has an error (yarn is very strict with memory limits)
* The code does not check if the requested resources (memory) actually exist. The whole thing will die with an exception if the user requests too much.
* The web interface is not started, since I currently do not extract the web-interface files on the application master. This easy to fix.
* The user can not send custom files with our jar. (But the user-jar can contain whatever the user wants).

Overall, this code is very user friendly: Basically one jar file to download. It should run with only one command line argument, everything else is extracted from the system

This implementation is similar to those of Spark and Tez: We allocate long-running containers. So you can submit multiple jobs and de-allocate the containers once everything is done.

I added this as "stratosphere-yarn" to stratosphere-addons. The work is based on https://github.com/hortonworks/simple-yarn-app.

Some maven stuff has been changed:
* the cloudera repos have been moved from the main pom to the `pact-hbase` package, since everything else is independent from cloudera versions
* the debian package has moved to a separate `debian-package` build profile, so the build should be a bit faster.
* The uberjar should be deployed automatically to `dopa.dima.tu-berlin.de`.