unit_6_launching_applications.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Unit 6 Launching Spark Applications"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Contents\n",
    "\n",
    "```\n",
    "6.1 Creating a Spark application\n",
    "6.2 Submitting an application to YARN\n",
    "6.3 Simple application submission example\n",
    "6.4 Cluster vs Client mode\n",
    "6.5 Dynamic resource allocation\n",
    "6.6 Adding dependencies\n",
    "6.7 How-to install additional Python packages\n",
    "6.8 Native Compression Libraries\n",
    "6.9 Submitting the application in the background\n",
    "6.10 Overriding configuration directory\n",
    "6.11 Complex application submission example\n",
    "6.12 Run an interactive shell\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Spark application\n",
    "An application is very similar to a notebook, but there are some minor changes that must be applied.\n",
    "\n",
    "The interactive notebook creates automatically the SparkContext (sc) and a SparkSession (spark) but in a standard application you must take care of creating them manually:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName('My Application') \\\n",
    "        .getOrCreate()\n",
    "    sc = spark.sparkContext\n",
    "    # ...\n",
    "    # Application specific code\n",
    "    # ..\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Submitting an application to YARN\n",
    "\n",
    "To submit an application to YARN you use the **spark-submit** utility:\n",
    "\n",
    "```\n",
    "spark-submit\n",
    "  --name NAME                 A name of your application.\n",
    "\n",
    "  --master MASTER_URL         spark://host:port, mesos://host:port, yarn, or local.\n",
    "  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally (\"client\") or\n",
    "                              on one of the worker machines inside the cluster (\"cluster\")\n",
    "                              (Default: client).\n",
    "  --queue QUEUE_NAME          The YARN queue to submit to (Default: \"default\").\n",
    "\n",
    "  --num-executors NUM         Number of executors to launch (Default: 2).\n",
    "  \n",
    "  --driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M).\n",
    "  --driver-cores NUM          Number of cores used by the driver, only in cluster mode\n",
    "                              (Default: 1).\n",
    "                              \n",
    "  --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).\n",
    "  --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode,\n",
    "                              or all available cores on the worker in standalone mode)\n",
    "```\n",
    "\n",
    "The main options to take into account for resource allocation are:\n",
    "\n",
    "* The `--num-executors` (spark.executor.instances as configuration property) option controls how many executors it will allocate for the application on the cluster .\n",
    "* The `--executor-memory` (spark.executor.memory configuration property) option controls the memory allocated per executor.\n",
    "* The `--executor-cores` (spark.executor.cores configuration property) option controls the cores allocated per executor.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Simple application submission example\n",
    "\n",
    "In the simplest case we can just use:\n",
    "\n",
    "    spark-submit test.py\n",
    "\n",
    "But in general we will frequently use:\n",
    "\n",
    "    spark-submit --deploy-mode cluster --conf spark.yarn.submit.waitAppCompletion=false test.py\n",
    "\n",
    "We can also give a name to the application (it will overwrite the name given in the python code) and we can explicitly indicate that we want to use yarn (which is the default in a Hadoop cluster):\n",
    "\n",
    "    spark-submit --master yarn --deploy-mode client --name testWC test.py\n",
    "    spark-submit --master yarn --deploy-mode cluster --name testWC test.py\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cluster vs Client mode\n",
    "\n",
    "### Client Mode\n",
    "![Client mode](https://bigdata.cesga.es/img/spark_client_mode.png)\n",
    "\n",
    "### Cluster Mode\n",
    "![Cluster mode](https://bigdata.cesga.es/img/spark_cluster_mode.png)\n",
    "\n",
    "More information: [Spark-on-YARN: Empower Spark Applications on Hadoop Cluster](https://www.slideshare.net/Hadoop_Summit/sparkonyarn-empower-spark-applications-on-hadoop-cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dynamic resource allocation\n",
    "Spark provides a mechanism to dynamically adjust the resources your application occupies based on the workload. This means that your application may give resources back to the cluster if they are no longer used and request them again later when there is demand. This feature is particularly useful if multiple applications share resources in your Spark cluster.\n",
    "\n",
    "Our cluster has this feature enabled so it **automatically expands new executors when they are needed**, instead of fixing them at launch time with --num-executors.\n",
    "\n",
    "This allows that interactive jobs dynamically add and remove executors during execution.\n",
    "\n",
    "It is important to notice that when you specify the `--num-executors` option without explicitly disabling dynamic resource allocation, then num-executors indicates the initial number of executors to allocate (by default it is 2).\n",
    "\n",
    "If you want to **disable dynamic resource allocation** and request a fixed number of executors you have to use the following spark-submit options:\n",
    "\n",
    "    spark-submit --conf spark.dynamicAllocation.enabled=false --num-executors 4 ..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding dependencies\n",
    "\n",
    "To add dependencies we have to distinguish two different cases, when we need to extend Spark code itself which is written in Scala, or when we need to add the of our Python code.\n",
    "\n",
    "When you need to add dependencies for Spark itself we can use:\n",
    "- The **--packages** option pulls directly the packages from the Central Maven Repository. This approach requires an internet connection.\n",
    "- The **--jars** option transfers associated jar files to the cluster.\n",
    "\n",
    "\n",
    "To include the dependencies of our Python program we can use:\n",
    "- The **--py-files** option adds .zip, .egg, or .py files to the PYTHONPATH."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adding Spark dependencies: packages\n",
    "You can add Spark dependencies directly using the maven coordinates:\n",
    "\n",
    "    spark-submit --packages com.databricks:spark-avro_2.10:2.0.1 ...\n",
    "    \n",
    "The usual place to look for plublic packages is the Maven Central Repository:\n",
    "\n",
    "    https://search.maven.org/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adding Spark dependencies: jar files\n",
    "You can also add exising jar files directly as dependencies:\n",
    "\n",
    "    spark-submit --jars /jar_path/spark-streaming-kafka-assembly_2.10-1.6.1.2.4.2.0-258.jar ..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adding Python dependencies: zip files\n",
    "The easiest way to add our Python dependencies is to package all the dependencies in a zip file.\n",
    "If our program is more than a simple script and it defines its own modules and packages, then it is also needed to package and distribute them so the executors can access them.\n",
    "\n",
    "If we have a `requirements.txt` file we can generate a `dependencies.zip` file including all the dependencies with the following commands:\n",
    "```\n",
    "pip install -t dependencies -r requirements.txt\n",
    "```\n",
    "\n",
    "If the package we need it is not in PyPI but we have its `setup.py` then we can generate easily a zip with it and its dependencies running from the directory where the setup.py is located:\n",
    "```\n",
    "pip install -t dependencies .\n",
    "```\n",
    "\n",
    "Then we just package all the dependencies in a zip file:\n",
    "```\n",
    "cd dependencies\n",
    "zip -r ../dependencies.zip .\n",
    "```\n",
    "\n",
    "And then we need to package also the code of our application:\n",
    "```\n",
    "zip -r my_program.zip my_program\n",
    "```\n",
    "\n",
    "Finally to submit your application you will use:\n",
    "\n",
    "    spark-submit --py-files dependencies.zip,my_program.zip ..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adding Python dependencies: egg files\n",
    "\n",
    "In case you have an egg file of a package you want to use, you can add it directly to the `--py-files` option of spark-submit or to the `sc.addPyFile()` method to make it available to the application. After that you can make use of it in your application in the standard way.\n",
    "\n",
    "    spark-submit --py-files /egg_path/avro-1.8.1-py2.7.egg ..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# First we add the egg file to the application environment\n",
    "sc.addPyFile('/home/cesga/jlopez/packages/ClusterShell-1.7.3-py2.7.egg')\n",
    "# Then we can import and use it in the standard way\n",
    "from ClusterShell.NodeSet import NodeSet\n",
    "nodeset = NodeSet('c[6601-6610]')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adding Python dependencies: wheel files\n",
    "Unfortunately wheel files are not yet supported.\n",
    "\n",
    "There was a feature request by it has been recently closed because no progress had been made since 2016:\n",
    "\n",
    "https://issues.apache.org/jira/browse/SPARK-6764"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How-to install additional Python packages\n",
    "The simplest way is to use **pip** with the `--user` option:\n",
    "\n",
    "    pip install --user pymongo\n",
    "    \n",
    "You can also create a **virtualenv** and use it to install all your dependencies:\n",
    "\n",
    "    virtualenv venv\n",
    "    . venv/bin/activate\n",
    "    \n",
    "In case you are using a virtualenv you have to point spark to the appropriate python interpreter for your virtualenv:\n",
    "\n",
    "    export PYSPARK_DRIVER_PYTHON=/home/cesga/jlopez/my_app/venv/bin/python\n",
    "    export PYSPARK_PYTHON=/home/cesga/jlopez/my_app/venv/bin/python\n",
    "\n",
    "You will also need to adjust the permissions of your HOME directory so the spark user can access this virtualenv.\n",
    "\n",
    "This a quick & dirty way that you can use during development but, for production, I would recommend the zip file alternative."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Native Compression Libraries\n",
    "\n",
    "To check native Hadoop and compression libraries availability you can run the `hadoop checknative` command:\n",
    "\n",
    "```\n",
    "[jlopez@cdh61-login6 ~]$ hadoop checknative\n",
    "...\n",
    "Native library checking:\n",
    "hadoop:  true /opt/cloudera/parcels/CDH-6.1.1-1.cdh6.1.1.p0.875250/lib/hadoop/lib/native/libhadoop.so.1.0.0\n",
    "zlib:    true /lib64/libz.so.1\n",
    "zstd  :  true /opt/cloudera/parcels/CDH-6.1.1-1.cdh6.1.1.p0.875250/lib/hadoop/lib/native/libzstd.so.1\n",
    "snappy:  true /opt/cloudera/parcels/CDH-6.1.1-1.cdh6.1.1.p0.875250/lib/hadoop/lib/native/libsnappy.so.1\n",
    "lz4:     true revision:10301\n",
    "bzip2:   true /lib64/libbz2.so.1\n",
    "openssl: true /lib64/libcrypto.so\n",
    "ISA-L:   true /opt/cloudera/parcels/CDH-6.1.1-1.cdh6.1.1.p0.875250/lib/hadoop/lib/native/libisal.so.2\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How-to submit the application in the background\n",
    "\n",
    "By default when you submit an application the spark-submit command keeps active waiting for application output. To avoid this behaviour use spark.yarn.submit.waitAppCompletion=false:\n",
    "\n",
    "    spark-submit --conf spark.yarn.submit.waitAppCompletion=false ...\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Overriding configuration directory\n",
    "To specify a different configuration directory other than the default “$SPARK_HOME/conf”, you can set SPARK_CONF_DIR. Spark will use the configuration files (spark-defaults.conf, spark-env.sh, log4j.properties, etc) from this directory.\n",
    "\n",
    "Example:\n",
    "\n",
    "    export SPARK_CONF_DIR=/home/cesga/jlopez/conf/\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Complex application submission example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here you can see a real example of how to submit a real application that consumes data from Kafka in avro format using Spark Streaming:\n",
    "\n",
    "```\n",
    "spark-submit --master yarn --deploy-mode cluster \\\n",
    "             --num-executors 2 \\\n",
    "             --conf spark.yarn.submit.waitAppCompletion=false  \\\n",
    "             --packages com.databricks:spark-avro_2.10:2.0.1 \\\n",
    "             --jars /home/cesga/jlopez/packages/spark-streaming-kafka-assembly_2.10-1.6.1.2.4.2.0-258.jar \\\n",
    "             --py-files /home/cesga/jlopez/packages/avro-1.8.1-py2.7.egg \\\n",
    "             --name 'SSH attack detector' \\             \n",
    "             ssh_attack_detector.py\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Running an interactive shell\n",
    "Additionally to Jupyter notebooks you can also use the command line interactive shell provided by Spark:\n",
    "\n",
    "    pyspark --master yarn --num-executors 4 --executor-cores 6 --queue interactive\n",
    "\n",
    "    --num-executors NUM         Number of executors to launch (Default: 2).\n",
    "    --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode)\n",
    "    --driver-cores NUM          Number of cores used by the driver, only in cluster mode (Default: 1).\n",
    "    --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).\n",
    "    --queue QUEUE_NAME          The YARN queue to submit to (Default: \"default\").    \n",
    "\n",
    "\n",
    "To use ipython instead of python for an interactive session use:\n",
    "\n",
    "    module load anaconda2\n",
    "    PYSPARK_DRIVER_PYTHON=$(which ipython) pyspark --queue interactive\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "* Exercise: In the solutions folder take a look at \"Unit_6_WordCount.py\" to see how the \"Unit 4 WordCount\" notebook was modified to create a spark application. Submit it to YARN.\n",
    "* Exercise: Modify the \"Unit 4 Working with meteorological data 2\" notebook and submit it to YARN.\n",
    "* Exercise: Modify the \"Unit 5 Working with meteorological data\" notebook and submit it to YARN."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}