Allow to specify hadoop minor version (2.4 and 2.6 at the moment)

README.md

@@ -197,6 +197,15 @@ EC2. These scripts are intended to be used by the default Spark AMI and is *not*
 expected to work on other AMIs. If you wish to start a cluster using Spark,
 please refer to http://spark-project.org/docs/latest/ec2-scripts.html 
 
+## Using S3 with Hadoop 2.6 or newer
+
+Starting Hadoop 2.6.0, s3 FS connector has been moved to a separate library called hadoop-aws. 
+
+- In order to make the package available add it as a dependency, `libraryDependencies += "org.apache.hadoop" % "hadoop-aws" % "2.6.4"`.
+- It can also be added it directly to spark-submit, `spark-submit --packages org.apache.hadoop:hadoop-aws:2.6.4 SimpleApp.py`.
+
+On other related note, it is recommended to use `s3a` and not `s3n` filesystem starting Hadoop 2.6.0.
+
 ## spark-ec2 Internals
 
 The Spark cluster setup is guided by the values set in `ec2-variables.sh`.`setup.sh`
@@ -237,3 +246,4 @@ after the templates have been configured. You can use the environment variables
 get a list of slave hostnames and `/root/spark-ec2/copy-dir` to sync a directory across machines.
 
 5. Modify `spark_ec2.py` to add your module to the list of enabled modules.
+

deploy.generic/root/spark-ec2/ec2-variables.sh

@@ -27,6 +27,7 @@ export MODULES="{{modules}}"
 export SPARK_VERSION="{{spark_version}}"
 export TACHYON_VERSION="{{tachyon_version}}"
 export HADOOP_MAJOR_VERSION="{{hadoop_major_version}}"
+export HADOOP_MINOR_VERSION="{{hadoop_minor_version}}"
 export SWAP_MB="{{swap}}"
 export SPARK_WORKER_INSTANCES="{{spark_worker_instances}}"
 export SPARK_MASTER_OPTS="{{spark_master_opts}}"

deploy_templates.py

@@ -73,6 +73,7 @@
   "spark_version": os.getenv("SPARK_VERSION"),
   "tachyon_version": os.getenv("TACHYON_VERSION"),
   "hadoop_major_version": os.getenv("HADOOP_MAJOR_VERSION"),
+  "hadoop_minor_version": os.getenv("HADOOP_MINOR_VERSION"),
   "java_home": os.getenv("JAVA_HOME"),
   "default_tachyon_mem": "%dMB" % tachyon_mb,
   "system_ram_mb": "%d" % system_ram_mb,

ephemeral-hdfs/init.sh

@@ -30,11 +30,27 @@ case "$HADOOP_MAJOR_VERSION" in
     cp /root/hadoop-native/* /root/ephemeral-hdfs/lib/native/
     ;;
   yarn)
-    wget http://s3.amazonaws.com/spark-related-packages/hadoop-2.4.0.tar.gz
-    echo "Unpacking Hadoop"
-    tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
-    rm hadoop-*.tar.gz
-    mv hadoop-2.4.0/ ephemeral-hdfs/
+    if [[ "$HADOOP_MINOR_VERSION" == "2.4" ]]; then
+      wget http://s3.amazonaws.com/spark-related-packages/hadoop-2.4.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.4.0/ ephemeral-hdfs/
+    elif [[ "$HADOOP_MINOR_VERSION" == "2.6" ]]; then
+      wget https://www.apache.org/dist/hadoop/core/hadoop-2.6.0/hadoop-2.6.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.6.0/ ephemeral-hdfs/
+    elif [[ "$HADOOP_MINOR_VERSION" == "2.7" ]]; then
+      wget https://www.apache.org/dist/hadoop/core/hadoop-2.7.0/hadoop-2.7.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.7.0/ ephemeral-hdfs/
+    else
+      echo "ERROR: Unknown Hadoop version"
+    fi
 
     # Have single conf dir
     rm -rf /root/ephemeral-hdfs/etc/hadoop/

persistent-hdfs/init.sh

@@ -29,11 +29,27 @@ case "$HADOOP_MAJOR_VERSION" in
     cp /root/hadoop-native/* /root/persistent-hdfs/lib/native/
     ;;
   yarn)
-    wget http://s3.amazonaws.com/spark-related-packages/hadoop-2.4.0.tar.gz
-    echo "Unpacking Hadoop"
-    tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
-    rm hadoop-*.tar.gz
-    mv hadoop-2.4.0/ persistent-hdfs/
+    if [[ "$HADOOP_MINOR_VERSION" == "2.4" ]]; then
+      wget http://s3.amazonaws.com/spark-related-packages/hadoop-2.4.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.4.0/ persistent-hdfs/
+    elif [[ "$HADOOP_MINOR_VERSION" == "2.6" ]]; then
+      wget https://www.apache.org/dist/hadoop/core/hadoop-2.6.0/hadoop-2.6.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.6.0/ persistent-hdfs/
+    elif [[ "$HADOOP_MINOR_VERSION" == "2.7" ]]; then
+      wget https://www.apache.org/dist/hadoop/core/hadoop-2.7.0/hadoop-2.7.0.tar.gz
+      echo "Unpacking Hadoop"
+      tar xvzf hadoop-*.tar.gz > /tmp/spark-ec2_hadoop.log
+      rm hadoop-*.tar.gz
+      mv hadoop-2.7.0/ persistent-hdfs/
+    else
+      echo "ERROR: Unknown Hadoop version"
+    fi
 
     # Have single conf dir
     rm -rf /root/persistent-hdfs/etc/hadoop/

scala/init.sh

@@ -11,10 +11,15 @@ SCALA_VERSION="2.10.3"
 
 if [[ "0.7.3 0.8.0 0.8.1" =~ $SPARK_VERSION ]]; then
   SCALA_VERSION="2.9.3"
+  wget http://s3.amazonaws.com/spark-related-packages/scala-$SCALA_VERSION.tgz
+elif [[ "2.0.0" =~ $SPARK_VERSION ]]; then
+  SCALA_VERSION="2.11.8"
+  wget http://downloads.lightbend.com/scala/2.11.8/scala-$SCALA_VERSION.tgz
+else
+  wget http://s3.amazonaws.com/spark-related-packages/scala-$SCALA_VERSION.tgz
 fi
 
 echo "Unpacking Scala"
-wget http://s3.amazonaws.com/spark-related-packages/scala-$SCALA_VERSION.tgz
 tar xvzf scala-*.tgz > /tmp/spark-ec2_scala.log
 rm scala-*.tgz
 mv `ls -d scala-* | grep -v ec2` scala

spark/init.sh

@@ -124,13 +124,71 @@ else
         wget http://s3.amazonaws.com/spark-related-packages/spark-1.2.1-bin-hadoop2.4.tgz
       fi
       ;;
+    1.2.2)
+      if [[ "$HADOOP_MAJOR_VERSION" == "1" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.2.2-bin-hadoop1.tgz
+      elif [[ "$HADOOP_MAJOR_VERSION" == "2" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.2.2-bin-cdh4.tgz
+      else
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.2.2-bin-hadoop2.4.tgz
+      fi
+      ;;
+    1.3.0)
+      if [[ "$HADOOP_MAJOR_VERSION" == "1" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.3.0-bin-hadoop1.tgz
+      elif [[ "$HADOOP_MAJOR_VERSION" == "2" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.3.0-bin-cdh4.tgz
+      else
+        wget http://s3.amazonaws.com/spark-related-packages/spark-1.3.0-bin-hadoop2.4.tgz
+      fi
+      ;;
+    2.0.0)
+      if [[ "$HADOOP_MAJOR_VERSION" == "1" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.0-bin-hadoop1.tgz
+      elif [[ "$HADOOP_MAJOR_VERSION" == "2" ]]; then
+        wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.0-bin-cdh4.tgz
+      else
+        if [[ "$HADOOP_MINOR_VERSION" == "2.4" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.0-bin-hadoop2.4.tgz
+        elif [[ "$HADOOP_MINOR_VERSION" == "2.6" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.0-bin-hadoop2.6.tgz
+        elif [[ "$HADOOP_MINOR_VERSION" == "2.7" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.0-bin-hadoop2.7.tgz
+        else
+          echo "ERROR: Unknown Hadoop version"
+        fi
+      fi
+      ;;
+    2.0.1)
+      if [[ "$HADOOP_MAJOR_VERSION" == "1" ]]; then
+          echo "ERROR: Not supported"
+      elif [[ "$HADOOP_MAJOR_VERSION" == "2" ]]; then
+          echo "ERROR: Not supported"
+      else
+        if [[ "$HADOOP_MINOR_VERSION" == "2.4" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.1-bin-hadoop2.4.tgz
+        elif [[ "$HADOOP_MINOR_VERSION" == "2.6" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.1-bin-hadoop2.6.tgz
+        elif [[ "$HADOOP_MINOR_VERSION" == "2.7" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-2.0.1-bin-hadoop2.7.tgz
+        else
+          echo "ERROR: Unknown Hadoop version"
+        fi
+      fi
+      ;;
     *)
       if [[ "$HADOOP_MAJOR_VERSION" == "1" ]]; then
         wget http://s3.amazonaws.com/spark-related-packages/spark-$SPARK_VERSION-bin-hadoop1.tgz
       elif [[ "$HADOOP_MAJOR_VERSION" == "2" ]]; then
         wget http://s3.amazonaws.com/spark-related-packages/spark-$SPARK_VERSION-bin-cdh4.tgz
       else
-        wget http://s3.amazonaws.com/spark-related-packages/spark-$SPARK_VERSION-bin-hadoop2.4.tgz
+        if [[ "$HADOOP_MINOR_VERSION" == "2.4" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-$SPARK_VERSION-bin-hadoop2.4.tgz
+        elif [[ "$HADOOP_MINOR_VERSION" == "2.6" ]]; then
+          wget http://s3.amazonaws.com/spark-related-packages/spark-$SPARK_VERSION-bin-hadoop2.6.tgz
+        else
+          echo "ERROR: Unknown Hadoop version"
+        fi
       fi
       if [ $? != 0 ]; then
         echo "ERROR: Unknown Spark version"

spark_ec2.py

@@ -80,6 +80,12 @@
     "2.0.0",
 ])
 
+VALID_HADOOP_MINOR_VERSIONS = set([
+    "2.4",
+    "2.5",
+    "2.7"
+])
+
 SPARK_TACHYON_MAP = {
     "1.0.0": "0.4.1",
     "1.0.1": "0.4.1",
@@ -240,6 +246,10 @@ def parse_args():
         "--hadoop-major-version", default="yarn",
         help="Major version of Hadoop. Valid options are 1 (Hadoop 1.0.4), 2 (CDH 4.2.0), yarn " +
              "(Hadoop 2.4.0) (default: %default)")
+    parser.add_option(
+        "--hadoop-minor-version", default="2.4",
+        help="Minor version of Hadoop. Valid options are 2.4 (Hadoop 2.4.0), 2.6 (Hadoop 2.6.0) and 2.7 (Hadoop 2.7.0). " +
+             "This only has any effect if yarn is specified as Hadoop major version/ (default: %default)")
     parser.add_option(
         "-D", metavar="[ADDRESS:]PORT", dest="proxy_port",
         help="Use SSH dynamic port forwarding to create a SOCKS proxy at " +
@@ -369,15 +379,19 @@ def get_or_make_group(conn, name, vpc_id):
         print("Creating security group " + name)
         return conn.create_security_group(name, "Spark EC2 group", vpc_id)
 
-def validate_spark_hadoop_version(spark_version, hadoop_version):
+def validate_spark_hadoop_version(spark_version, hadoop_version, hadoop_minor_version):
     if "." in spark_version:
         parts = spark_version.split(".")
         if parts[0].isdigit():
             spark_major_version = float(parts[0])
             if spark_major_version > 1.0 and hadoop_version != "yarn":
-              print("Spark version: {v}, does not support Hadoop version: {hv}".
+              print("Spark version: {v}, does not support Hadoop major version: {hv}".
                     format(v=spark_version, hv=hadoop_version), file=stderr)
               sys.exit(1)
+            if hadoop_version == "yarn" and hadoop_minor_version not in VALID_HADOOP_MINOR_VERSIONS:
+              print("Spark version: {v}, does not support Hadoop minor version: {hm}, supported minor versions: {sv}".
+                    format(v=spark_version, hm=hadoop_minor_version, sv=",".join(VALID_HADOOP_MINOR_VERSIONS)), file=stderr)
+              sys.exit(1)
         else:
             print("Invalid Spark version: {v}".format(v=spark_version), file=stderr)
             sys.exit(1)
@@ -1084,7 +1098,7 @@ def deploy_files(conn, root_dir, opts, master_nodes, slave_nodes, modules):
     if "." in opts.spark_version:
         # Pre-built Spark deploy
         spark_v = get_validate_spark_version(opts.spark_version, opts.spark_git_repo)
-        validate_spark_hadoop_version(spark_v, opts.hadoop_major_version)
+        validate_spark_hadoop_version(spark_v, opts.hadoop_major_version, opts.hadoop_minor_version)
         tachyon_v = get_tachyon_version(spark_v)
     else:
         # Spark-only custom deploy
@@ -1111,6 +1125,7 @@ def deploy_files(conn, root_dir, opts, master_nodes, slave_nodes, modules):
         "spark_version": spark_v,
         "tachyon_version": tachyon_v,
         "hadoop_major_version": opts.hadoop_major_version,
+        "hadoop_minor_version": opts.hadoop_minor_version,
         "spark_worker_instances": worker_instances_str,
         "spark_master_opts": opts.master_opts
     }
@@ -1295,7 +1310,7 @@ def real_main():
 
     # Input parameter validation
     spark_v = get_validate_spark_version(opts.spark_version, opts.spark_git_repo)
-    validate_spark_hadoop_version(spark_v, opts.hadoop_major_version)
+    validate_spark_hadoop_version(spark_v, opts.hadoop_major_version, opts.hadoop_minor_version)
 
     if opts.wait is not None:
         # NOTE: DeprecationWarnings are silent in 2.7+ by default.

tachyon/init.sh

@@ -13,6 +13,9 @@ then
   # Not yet supported
   echo "Tachyon git hashes are not yet supported. Please specify a Tachyon release version."
 # Pre-package tachyon version
+if [[ "$HADOOP_MAJOR_VERSION" == "yarn" || "$SPARK_VERSION" > "2" ]]
+then
+  echo "Tachyon is not supported with yarn or Spark 2.0.0 and newer."
 else
   case "$TACHYON_VERSION" in
     0.3.0)