Repository: nchammas/flintrock (python)

Issue:
IAM roles do not seem to allow Spark to access S3 files without explicit credentials
Flintrock: so fast! I'm super excited. I typically start launching a cluster from home, take the bus to work, and if I'm lucky, my cluster will be running when I get there. You write some pretty python too!

One nice feature of spark-ec2 is that it can copy AWS credentials and is setup (almost, have to export some things) right away to access data directly from s3. Since the data I use is exclusively in s3, this is a huge help. 

I don't have an obvious solution. I have a functioning core-site.xml that allows me to use hadoop from the command line, although I had to symlink /etc/hadoop to ~/hadoop/etc/hadoop. But that configuration doesn't seem to be recognized once I'm running spark, i.e. sc.textFile("s3://somewhere/somefile").first() raises an error that there is no s3/s3n framework. I am just hacking around, and I don't really know my way around the guts of hadoop (there are two! core-site.xml files, which do I use?), so there is surely a better way. 

I'm willing to contribute code/PRs too if someone can point me in the right direction!


Pull request:
Enable easy S3 access
Lots of people have trouble accessing S3 from their Flintrock clusters.

#90, which is about accessing S3 from Flintrock clusters, is the most visited issue on this project. A related issue, #88, which is driven by the same problem, is the second-most visited issue on this project. Two recent guides that go over how to use Flintrock -- [this one](https://github.com/PiercingDan/spark-Jupyter-AWS/) and [this one](https://alexioannides.com/2016/08/18/building-a-data-science-platform-for-rd-part-2-deploying-spark-on-aws-using-flintrock/) -- take time to address the same issue.

This PR attempts to address this common problem by 1) setting better defaults that enable Spark on Flintrock clusters to seamlessly access data on S3, and 2) by providing instructions in the README on how to make use of these new defaults.

I tested this PR by launching several clusters in a variety of configurations. I was able to seamlessly access S3 in all cases.

It seems to be working well, but I would like to get some feedback from people who have hit this issue in the past to make sure I'm headed in the right direction here:

* cc @PiercingDan and @AlexIoannides, who recently authored some guides that cover this issue in Flintrock.
* cc @marcuscollins, @hhbyyh, and @ereed-tesla, who reported this issue or discussed it in #90 and #88.

I know this PR may be too late for some of you, since you may have moved on or come up with your own workaround. So no hard feelings if you are not interested. And of course, if anyone else reading this would like to chime in with their feedback that would also be helpful.

If you would like to install Flintrock directly from this PR (assuming you are running Python 3.4+), you can do that with this:

```sh
pip install git+https://github.com/nchammas/flintrock@easy-s3-access
```

Fixes #90.


Patch:
diff --git a/.travis.yml b/.travis.yml
index c7d3638..72a7864 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,6 +8,7 @@ install:
 script:
   - "py.test ./tests/test_static.py"
   - "py.test ./tests/test_flintrock.py"
+  - "py.test ./tests/test_core.py"
   - "pip install -r requirements/maintainer.pip"
   - "py.test ./tests/test_pyinstaller_packaging.py"
 addons:
diff --git a/README.md b/README.md
index cfb41ed..abf3393 100644
--- a/README.md
+++ b/README.md
@@ -18,10 +18,10 @@ Here's a quick way to launch a cluster on EC2, assuming you already have an [AWS
 ```sh
 flintrock launch test-cluster \
     --num-slaves 1 \
-    --spark-version 2.0.2 \
+    --spark-version 2.1.0 \
     --ec2-key-name key_name \
     --ec2-identity-file /path/to/key.pem \
-    --ec2-ami ami-b73b63a0 \
+    --ec2-ami ami-9be6f38c \
     --ec2-user ec2-user
 ```
 
@@ -57,6 +57,30 @@ flintrock <subcommand> --help
 
 That's not all. Flintrock has a few more [features](#features) that you may find interesting.
 
+### Accessing data on S3
+
+We recommend you access data on S3 from your Flintrock cluster by following
+these steps:
+
+1. Setup an [IAM Role](http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html)
+   that grants access to S3 as desired. Reference this role when you launch
+   your cluster using the `--ec2-instance-profile-name` option (or its
+   equivalent in your `config.yaml` file).
+2. Reference S3 paths in your Spark code using the `s3a://` prefix. `s3a://` is
+   backwards compatible with `s3n://` and replaces both `s3n://` and `s3://`.
+   The Hadoop project [recommends using `s3a://`](https://hadoop.apache.org/docs/current/hadoop-aws/tools/hadoop-aws/index.html#S3A)
+   since it is actively developed, supports larger files, and offers
+   better performance.
+3. Make sure Flintrock is configured to use Hadoop/HDFS 2.7+. Earlier
+   versions of Hadoop do not have solid implementations of `s3a://`.
+   Flintrock's default is Hadoop 2.7.3, so you don't need to do anything
+   here if you're using a vanilla configuration.
+
+With this approach you don't need to copy around your AWS credentials
+or pass them into your Spark programs. As long as the assigned IAM role
+allows it, Spark will be able to read and write data to S3 simply by
+referencing the appropriate path (e.g. `s3a://bucket/path/to/file`).
+
 
 ## Installation
 
@@ -196,7 +220,7 @@ provider: ec2
 
 services:
   spark:
-    version: 2.0.2
+    version: 2.1.0
 
 launch:
   num-slaves: 1
@@ -207,7 +231,7 @@ providers:
     identity-file: /path/to/.ssh/key.pem
     instance-type: m3.medium
     region: us-east-1
-    ami: ami-b73b63a0
+    ami: ami-9be6f38c
     user: ec2-user
 ```
 
diff --git a/flintrock/config.yaml.template b/flintrock/config.yaml.template
index 1767c40..29edbe6 100644
--- a/flintrock/config.yaml.template
+++ b/flintrock/config.yaml.template
@@ -1,6 +1,6 @@
 services:
   spark:
-    version: 2.0.2
+    version: 2.1.0
     # git-commit: latest  # if not 'latest', provide a full commit SHA; e.g. d6dc12ef0146ae409834c78737c116050961f350
     # git-repository:  # optional; defaults to https://github.com/apache/spark
     # optional; defaults to download from from the official Spark S3 bucket
@@ -24,7 +24,7 @@ providers:
     instance-type: m3.medium
     region: us-east-1
     # availability-zone: <name>
-    ami: ami-b73b63a0   # Amazon Linux, us-east-1
+    ami: ami-9be6f38c   # Amazon Linux, us-east-1
     user: ec2-user
     # ami: ami-61bbf104   # CentOS 7, us-east-1
     # user: centos
diff --git a/flintrock/core.py b/flintrock/core.py
index 19e8ccf..20dc79b 100644
--- a/flintrock/core.py
+++ b/flintrock/core.py
@@ -423,30 +423,65 @@ class FlintrockCluster:
             user=user,
             identity_file=identity_file)
 
-    def generate_template_mapping(self, *, service: str) -> dict:
-        """
-        Generate a template mapping from a FlintrockCluster instance that we can use
-        to fill in template parameters.
-        """
-        root_dir = posixpath.join(self.storage_dirs.root, service)
-        ephemeral_dirs = ','.join(posixpath.join(path, service) for path in self.storage_dirs.ephemeral)
-
-        template_mapping = {
-            'master_ip': self.master_ip,
-            'master_host': self.master_host,
-            'slave_ips': '\n'.join(self.slave_ips),
-            'slave_hosts': '\n'.join(self.slave_hosts),
-            'root_dir': root_dir,
-            'ephemeral_dirs': ephemeral_dirs,
-
-            # If ephemeral storage is available, it replaces the root volume, which is
-            # typically persistent. We don't want to mix persistent and ephemeral
-            # storage since that causes problems after cluster stop/start; some volumes
-            # have leftover data, whereas others start fresh.
-            'root_ephemeral_dirs': ephemeral_dirs if ephemeral_dirs else root_dir,
-        }
 
-        return template_mapping
+def generate_template_mapping(
+    *,
+    cluster: FlintrockCluster,
+    # If we add additional services later on we may want to refactor
+    # this to take a list of services and dynamically pull the service
+    # name.
+    hadoop_version: str,
+    spark_version: str
+) -> dict:
+    """
+    Generate a template mapping from a FlintrockCluster instance that we can use
+    to fill in template parameters.
+    """
+    hadoop_root_dir = posixpath.join(cluster.storage_dirs.root, 'hadoop')
+    hadoop_ephemeral_dirs = ','.join(
+        posixpath.join(path, 'hadoop')
+        for path in cluster.storage_dirs.ephemeral
+    )
+    spark_root_dir = posixpath.join(cluster.storage_dirs.root, 'spark')
+    spark_ephemeral_dirs = ','.join(
+        posixpath.join(path, 'spark')
+        for path in cluster.storage_dirs.ephemeral
+    )
+
+    template_mapping = {
+        'master_ip': cluster.master_ip,
+        'master_host': cluster.master_host,
+        'slave_ips': '\n'.join(cluster.slave_ips),
+        'slave_hosts': '\n'.join(cluster.slave_hosts),
+
+        'hadoop_version': hadoop_version,
+        'hadoop_short_version': '.'.join(hadoop_version.split('.')[:2]),
+        'spark_version': spark_version,
+        'spark_short_version': '.'.join(spark_version.split('.')[:2]),
+
+        'hadoop_root_dir': hadoop_root_dir,
+        'hadoop_ephemeral_dirs': hadoop_ephemeral_dirs,
+        'spark_root_dir': spark_root_dir,
+        'spark_ephemeral_dirs': spark_ephemeral_dirs,
+
+        # If ephemeral storage is available, it replaces the root volume, which is
+        # typically persistent. We don't want to mix persistent and ephemeral
+        # storage since that causes problems after cluster stop/start; some volumes
+        # have leftover data, whereas others start fresh.
+        'hadoop_root_ephemeral_dirs': hadoop_ephemeral_dirs if hadoop_ephemeral_dirs else hadoop_root_dir,
+        'spark_root_ephemeral_dirs': spark_ephemeral_dirs if spark_ephemeral_dirs else spark_root_dir,
+    }
+
+    return template_mapping
+
+
+# TODO: Cache these files. (?) They are being read potentially tens or
+#       hundreds of times. Maybe it doesn't matter because the files
+#       are so small.
+def get_formatted_template(*, path: str, mapping: dict) -> str:
+    with open(path) as f:
+        formatted = f.read().format(**mapping)
+    return formatted
 
 
 def run_against_hosts(*, partial_func: functools.partial, hosts: list):
diff --git a/flintrock/flintrock.py b/flintrock/flintrock.py
index aece38e..58cb9d6 100644
--- a/flintrock/flintrock.py
+++ b/flintrock/flintrock.py
@@ -187,13 +187,14 @@ def cli(cli_context, config, provider):
 @click.argument('cluster-name')
 @click.option('--num-slaves', type=click.IntRange(min=1), required=True)
 @click.option('--install-hdfs/--no-install-hdfs', default=False)
-@click.option('--hdfs-version')
+@click.option('--hdfs-version', default='2.7.3')
 @click.option('--hdfs-download-source',
               help="URL to download Hadoop from.",
               default='http://www.apache.org/dyn/closer.lua/hadoop/common/hadoop-{v}/hadoop-{v}.tar.gz?as_json',
               show_default=True)
 @click.option('--install-spark/--no-install-spark', default=True)
 @click.option('--spark-version',
+              default='2.1.0',
               help="Spark release version to install.")
 @click.option('--spark-download-source',
               help="URL to download a release of Spark from.",
@@ -314,7 +315,11 @@ def launch(
         services += [hdfs]
     if install_spark:
         if spark_version:
-            spark = Spark(version=spark_version, download_source=spark_download_source)
+            spark = Spark(
+                version=spark_version,
+                hadoop_version=hdfs_version,
+                download_source=spark_download_source,
+            )
         elif spark_git_commit:
             print(
                 "Warning: Building Spark takes a long time. "
@@ -324,7 +329,9 @@ def launch(
                 print("Building Spark at latest commit: {c}".format(c=spark_git_commit))
             spark = Spark(
                 git_commit=spark_git_commit,
-                git_repository=spark_git_repository)
+                git_repository=spark_git_repository,
+                hadoop_version=hdfs_version,
+            )
         services += [spark]
 
     if provider == 'ec2':
diff --git a/flintrock/services.py b/flintrock/services.py
index 5e13608..f850d3e 100644
--- a/flintrock/services.py
+++ b/flintrock/services.py
@@ -9,7 +9,11 @@ import urllib.request
 import paramiko
 
 # Flintrock modules
-from .core import FlintrockCluster
+from .core import (
+    FlintrockCluster,
+    generate_template_mapping,
+    get_formatted_template,
+)
 from .ssh import ssh_check_output
 
 FROZEN = getattr(sys, 'frozen', False)
@@ -22,15 +26,6 @@ else:
 SCRIPTS_DIR = os.path.join(THIS_DIR, 'scripts')
 
 
-# TODO: Cache these files. (?) They are being read potentially tens or
-#       hundreds of times. Maybe it doesn't matter because the files
-#       are so small.
-def get_formatted_template(path: str, mapping: dict) -> str:
-    with open(path) as f:
-        formatted = f.read().format(**mapping)
-    return formatted
-
-
 class FlintrockService:
     """
     This is an abstract class. Implementations of this class capture all the logic
@@ -108,7 +103,7 @@ class FlintrockService:
 
 
 class HDFS(FlintrockService):
-    def __init__(self, version, download_source):
+    def __init__(self, *, version, download_source):
         self.version = version
         self.download_source = download_source
         self.manifest = {'version': version, 'download_source': download_source}
@@ -154,7 +149,8 @@ class HDFS(FlintrockService):
             'hadoop/conf/slaves',
             'hadoop/conf/hadoop-env.sh',
             'hadoop/conf/core-site.xml',
-            'hadoop/conf/hdfs-site.xml']
+            'hadoop/conf/hdfs-site.xml',
+        ]
 
         for template_path in template_paths:
             ssh_check_output(
@@ -165,7 +161,13 @@ class HDFS(FlintrockService):
                     f=shlex.quote(
                         get_formatted_template(
                             path=os.path.join(THIS_DIR, "templates", template_path),
-                            mapping=cluster.generate_template_mapping(service='hdfs'))),
+                            mapping=generate_template_mapping(
+                                cluster=cluster,
+                                hadoop_version=self.version,
+                                # Hadoop doesn't need to know what
+                                # Spark version we're using.
+                                spark_version='',
+                            ))),
                     p=shlex.quote(template_path)))
 
     # TODO: Convert this into start_master() and split master- or slave-specific
@@ -203,8 +205,15 @@ class HDFS(FlintrockService):
 
 
 class Spark(FlintrockService):
-    def __init__(self, version: str=None, download_source: str=None,
-                 git_commit: str=None, git_repository: str=None):
+    def __init__(
+        self,
+        *,
+        version: str=None,
+        hadoop_version: str,
+        download_source: str=None,
+        git_commit: str=None,
+        git_repository: str=None
+    ):
         # TODO: Convert these checks into something that throws a proper exception.
         #       Perhaps reuse logic from CLI.
         assert bool(version) ^ bool(git_commit)
@@ -212,12 +221,14 @@ class Spark(FlintrockService):
             assert git_repository
 
         self.version = version
+        self.hadoop_version = hadoop_version
         self.download_source = download_source
         self.git_commit = git_commit
         self.git_repository = git_repository
 
         self.manifest = {
             'version': version,
+            'hadoop_version': hadoop_version,
             'download_source': download_source,
             'git_commit': git_commit,
             'git_repository': git_repository}
@@ -261,13 +272,15 @@ class Spark(FlintrockService):
                         cd spark
                         git reset --hard {commit}
                         if [ -e "make-distribution.sh" ]; then
-                            ./make-distribution.sh -Phadoop-2.6
+                            ./make-distribution.sh -Phadoop-{hadoop_short_version}
                         else
-                            ./dev/make-distribution.sh -Phadoop-2.6
+                            ./dev/make-distribution.sh -Phadoop-{hadoop_short_version}
                         fi
                     """.format(
                         repo=shlex.quote(self.git_repository),
-                        commit=shlex.quote(self.git_commit)))
+                        commit=shlex.quote(self.git_commit),
+                        hadoop_short_version='.'.join(self.hadoop_version.split('.')[:2]),
+                    ))
             ssh_check_output(
                 client=ssh_client,
                 command="""
@@ -289,7 +302,9 @@ class Spark(FlintrockService):
             cluster: FlintrockCluster):
         template_paths = [
             'spark/conf/spark-env.sh',
-            'spark/conf/slaves']
+            'spark/conf/slaves',
+            'spark/conf/spark-defaults.conf',
+        ]
         for template_path in template_paths:
             ssh_check_output(
                 client=ssh_client,
@@ -299,7 +314,11 @@ class Spark(FlintrockService):
                     f=shlex.quote(
                         get_formatted_template(
                             path=os.path.join(THIS_DIR, "templates", template_path),
-                            mapping=cluster.generate_template_mapping(service='spark'))),
+                            mapping=generate_template_mapping(
+                                cluster=cluster,
+                                hadoop_version=self.hadoop_version,
+                                spark_version=self.version,
+                            ))),
                     p=shlex.quote(template_path)))
 
     # TODO: Convert this into start_master() and split master- or slave-specific
diff --git a/flintrock/templates/hadoop/conf/core-site.xml b/flintrock/templates/hadoop/conf/core-site.xml
index d53ffbd..b4bebb4 100644
--- a/flintrock/templates/hadoop/conf/core-site.xml
+++ b/flintrock/templates/hadoop/conf/core-site.xml
@@ -4,7 +4,7 @@
 <configuration>
   <property>
     <name>hadoop.tmp.dir</name>
-    <value>{root_ephemeral_dirs}</value>
+    <value>{hadoop_root_ephemeral_dirs}</value>
   </property>
 
   <property>
diff --git a/flintrock/templates/hadoop/conf/hdfs-site.xml b/flintrock/templates/hadoop/conf/hdfs-site.xml
index 94e34eb..c2c117c 100644
--- a/flintrock/templates/hadoop/conf/hdfs-site.xml
+++ b/flintrock/templates/hadoop/conf/hdfs-site.xml
@@ -9,6 +9,6 @@
 
   <property>
     <name>dfs.datanode.data.dir</name>
-    <value>{root_ephemeral_dirs}</value>
+    <value>{hadoop_root_ephemeral_dirs}</value>
   </property>
 </configuration>
diff --git a/flintrock/templates/spark/conf/spark-defaults.conf b/flintrock/templates/spark/conf/spark-defaults.conf
new file mode 100644
index 0000000..b4ed251
--- /dev/null
+++ b/flintrock/templates/spark/conf/spark-defaults.conf
@@ -0,0 +1,1 @@
+spark.jars.packages    org.apache.hadoop:hadoop-aws:{hadoop_version}
diff --git a/flintrock/templates/spark/conf/spark-env.sh b/flintrock/templates/spark/conf/spark-env.sh
index d849910..77bd7f8 100644
--- a/flintrock/templates/spark/conf/spark-env.sh
+++ b/flintrock/templates/spark/conf/spark-env.sh
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-export SPARK_LOCAL_DIRS="{root_ephemeral_dirs}"
+export SPARK_LOCAL_DIRS="{spark_root_ephemeral_dirs}"
 
 # Standalone cluster options
 export SPARK_EXECUTOR_INSTANCES="1"


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 92bfaa9..36fa8a5 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -4,14 +4,40 @@ import tempfile
 import uuid
 from collections import OrderedDict
 
-# External modules
+# Flintrock
+from flintrock.core import FlintrockCluster, StorageDirs
+
+# External
 import pytest
 
 HADOOP_VERSION = '2.7.3'
-SPARK_VERSION = '2.0.2'
+SPARK_VERSION = '2.1.0'
 SPARK_GIT_COMMIT = '584354eaac02531c9584188b143367ba694b0c34'  # 2.0.2
 
 
+class Dummy():
+    pass
+
+
+@pytest.fixture(scope='session')
+def dummy_cluster():
+    storage_dirs = StorageDirs(
+        root='/media/root',
+        ephemeral=['/media/eph1', '/media/eph2'],
+        persistent=None,
+    )
+
+    cluster = Dummy()
+    cluster.name = 'test'
+    cluster.storage_dirs = storage_dirs
+    cluster.master_ip = '10.0.0.1'
+    cluster.master_host = 'master.hostname'
+    cluster.slave_ips = ['10.0.0.2']
+    cluster.slave_hosts = ['slave1.hostname']
+
+    return cluster
+
+
 def random_string():
     return str(uuid.uuid4())[:8]
 
diff --git a/tests/test_core.py b/tests/test_core.py
new file mode 100644
index 0000000..56e58ed
--- /dev/null
+++ b/tests/test_core.py
@@ -0,0 +1,29 @@
+import os
+
+# Flintrock
+from flintrock.core import (
+    generate_template_mapping,
+    get_formatted_template,
+)
+
+FLINTROCK_ROOT_DIR = (
+    os.path.dirname(
+        os.path.dirname(
+            os.path.realpath(__file__))))
+
+
+def test_templates(dummy_cluster):
+    template_dir = os.path.join(FLINTROCK_ROOT_DIR, 'flintrock', 'templates')
+    for (dirpath, dirnames, filenames) in os.walk(template_dir):
+        if filenames:
+            for filename in filenames:
+                template_path = os.path.join(dirpath, filename)
+                mapping = generate_template_mapping(
+                    cluster=dummy_cluster,
+                    hadoop_version='',
+                    spark_version='',
+                )
+                get_formatted_template(
+                    path=template_path,
+                    mapping=mapping,
+                )