Repository: nchammas/flintrock
Pull request #181 (2017-01-26 23:50:21)

Added --spark-executor-instances to specify how many instances by worker
This PR makes the SPARK_EXECUTOR_INSTANCES be configurable.

I tested this PR by manually launching a cluster and running a job.

nchammas: What's the motivation for adding this?

I am wary about adding more options for Spark since there is a potentially [a lot of stuff](https://github.com/apache/spark/blob/master/conf/spark-env.sh.template) we could end up duplicating into Flintrock. Typically, I'd just say "Use `run-command` to deploy whatever config changes you want to the cluster." It's clumsy but generic.

Regarding executor instances in particular, why not vary your instance size or count if you want to control how many executors you have?
douglaz: @nchammas I'm sorry, but I'm so used to this feature that I don't even know how people do without it. I mean, how can you run the same job in a `x1.32xlarge` and a `r3.xlarge` without using this option or, alternatively, tweaking other JVM/Spark parameters?

Our solution is to break any big machine (big machine is any machine with more than 60GB of ram) in small machines (JVMs) with ~30GB. So in this example, we will set the x1.32xlarge to 64 executor instances. In this way we always have a homogenous cluster from the perspective of executor JVM, letting us use the same GC defaults for any configuration. Also, even considering the case where no GC tuning is needed, like when the machine is an r3.2xlarge, multiple smaller JVMs seems to run faster than a single big JVM (e.g networking/data transfer is improved).

So this simple single option let us avoid the need to use many other configurations and handling different machines in different ways.
douglaz: @nchammas rewritten to conform with current master
nchammas: Thanks for sharing the additional information. Makes sense to me. I'm curious if @BenFradet, @serialx, or any other users using Flintrock with large instance types have felt the need to use this setting before.

I'll try to code review this PR next, hopefully by next weekend (though, as with most open source, I can't make any promises).
serialx: Code seems to be okay to me. `export SPARK_EXECUTOR_CORES="$(($(nproc) / {spark_executor_instances}))"` could be a reasonable default.

It is [recommended to set `SPARK_EXECUTOR_INSTANCES`][1] when the instance has many cores.

I've also used to set `SPARK_EXECUTOR_INSTANCES` greater than `2` when we used spark-ec2 script. But resorted to using `SPARK_EXECUTOR_INSTANCES=1` after we migrated to flintrock. It would be good to support this feature in flintrock. This settings also makes sense when doing the benchmarks too.

[1]: https://www.slideshare.net/AmazonWebServices/bdt309-data-science-best-practices-for-apache-spark-on-amazon-emr/41?src=clipshare

Patch:
diff --git a/flintrock/config.yaml.template b/flintrock/config.yaml.template
index 86619f9..43bba71 100644
--- a/flintrock/config.yaml.template
+++ b/flintrock/config.yaml.template
@@ -8,6 +8,7 @@ services:
     #   - Spark must be pre-built
     #   - must be a tar.gz file
     # download-source: "https://www.example.com/files/spark/{v}/spark-{v}.tar.gz"
+    # executor-instances: 1
   hdfs:
     version: 2.7.3
     # optional; defaults to download from a dynamically selected Apache mirror
diff --git a/flintrock/core.py b/flintrock/core.py
index 15fa389..535285b 100644
--- a/flintrock/core.py
+++ b/flintrock/core.py
@@ -434,6 +434,7 @@ def generate_template_mapping(
     # 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.
+    spark_executor_instances: int,
     hadoop_version: str,
     spark_version: str
 ) -> dict:
@@ -463,6 +464,8 @@ def generate_template_mapping(
         'spark_version': spark_version,
         'spark_short_version': '.'.join(spark_version.split('.')[:2]) if '.' in spark_version else spark_version,
 
+        'spark_executor_instances': spark_executor_instances,
+
         'hadoop_root_dir': hadoop_root_dir,
         'hadoop_ephemeral_dirs': hadoop_ephemeral_dirs,
         'spark_root_dir': spark_root_dir,
diff --git a/flintrock/flintrock.py b/flintrock/flintrock.py
index 7c57bdb..0494849 100644
--- a/flintrock/flintrock.py
+++ b/flintrock/flintrock.py
@@ -219,6 +219,8 @@ def cli(cli_context, config, provider, debug):
               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-executor-instances', default=1,
+              help="How many executor instances per worker.")
 @click.option('--spark-version',
               # Don't set a default here because it may conflict with
               # the config file.
@@ -280,6 +282,7 @@ def launch(
         hdfs_version,
         hdfs_download_source,
         install_spark,
+        spark_executor_instances,
         spark_version,
         spark_git_commit,
         spark_git_repository,
@@ -353,6 +356,7 @@ def launch(
     if install_spark:
         if spark_version:
             spark = Spark(
+                spark_executor_instances=spark_executor_instances,
                 version=spark_version,
                 hadoop_version=hdfs_version,
                 download_source=spark_download_source,
@@ -365,6 +369,7 @@ def launch(
                 spark_git_commit = get_latest_commit(spark_git_repository)
                 logger.info("Building Spark at latest commit: {c}".format(c=spark_git_commit))
             spark = Spark(
+                spark_executor_instances=spark_executor_instances,
                 git_commit=spark_git_commit,
                 git_repository=spark_git_repository,
                 hadoop_version=hdfs_version,
diff --git a/flintrock/services.py b/flintrock/services.py
index 308424c..8515ed1 100644
--- a/flintrock/services.py
+++ b/flintrock/services.py
@@ -171,6 +171,7 @@ class HDFS(FlintrockService):
                                 # Hadoop doesn't need to know what
                                 # Spark version we're using.
                                 spark_version='',
+                                spark_executor_instances=0,
                             ))),
                     p=shlex.quote(template_path)))
 
@@ -212,6 +213,7 @@ class Spark(FlintrockService):
     def __init__(
         self,
         *,
+        spark_executor_instances: int,
         version: str=None,
         hadoop_version: str,
         download_source: str=None,
@@ -224,6 +226,7 @@ class Spark(FlintrockService):
         if git_commit:
             assert git_repository
 
+        self.spark_executor_instances = spark_executor_instances
         self.version = version
         self.hadoop_version = hadoop_version
         self.download_source = download_source
@@ -232,6 +235,7 @@ class Spark(FlintrockService):
 
         self.manifest = {
             'version': version,
+            'spark_executor_instances': spark_executor_instances,
             'hadoop_version': hadoop_version,
             'download_source': download_source,
             'git_commit': git_commit,
@@ -304,6 +308,7 @@ class Spark(FlintrockService):
             self,
             ssh_client: paramiko.client.SSHClient,
             cluster: FlintrockCluster):
+
         template_paths = [
             'spark/conf/spark-env.sh',
             'spark/conf/slaves',
@@ -320,6 +325,7 @@ class Spark(FlintrockService):
                             path=os.path.join(THIS_DIR, "templates", template_path),
                             mapping=generate_template_mapping(
                                 cluster=cluster,
+                                spark_executor_instances=self.spark_executor_instances,
                                 hadoop_version=self.hadoop_version,
                                 spark_version=self.version or self.git_commit,
                             ))),
diff --git a/flintrock/templates/spark/conf/spark-env.sh b/flintrock/templates/spark/conf/spark-env.sh
index 968e659..4e7b096 100644
--- a/flintrock/templates/spark/conf/spark-env.sh
+++ b/flintrock/templates/spark/conf/spark-env.sh
@@ -3,7 +3,8 @@
 export SPARK_LOCAL_DIRS="{spark_root_ephemeral_dirs}"
 
 # Standalone cluster options
-export SPARK_EXECUTOR_INSTANCES="1"
+export SPARK_EXECUTOR_INSTANCES="{spark_executor_instances}"
+export SPARK_EXECUTOR_CORES="$(($(nproc) / {spark_executor_instances}))"
 export SPARK_WORKER_CORES="$(nproc)"
 
 export SPARK_MASTER_HOST="{master_host}"


Test patch:
diff --git a/tests/test_core.py b/tests/test_core.py
index d2699eb..09443c4 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -29,6 +29,7 @@ def test_templates(dummy_cluster, spark_version):
                     cluster=dummy_cluster,
                     hadoop_version='',
                     spark_version=spark_version,
+                    spark_executor_instances=0,
                 )
                 get_formatted_template(
                     path=template_path,