07 - Ejecucion de un programa Spark y montaje de clusters.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![RDD key pair](media/10.spark_run_scripts.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ejecución de un programa Spark y montaje de clusters\n",
    "\n",
    "#### Comando `spark-submit`\n",
    "\n",
    "-   Permite lanzar programas Spark a un cluster\n",
    "\n",
    "-   Ejemplo:\n",
    "```sh\n",
    "$ bin/spark-submit --master yarn --deploy-mode cluster \\  \n",
    "     --py-files otralib.zip,otrofich.py \\  \n",
    "     --num-executors 10 --executor-cores 2 \\  \n",
    "     mi-script.py opciones_del_script\n",
    "```\n",
    "\n",
    "#### Opciones de `spark-submit`\n",
    "\n",
    "-   `master`: cluster manager a usar (opciones: `yarn`, `mesos://host:port`, `spark://host:port`, `local[n]`)\n",
    "\n",
    "-   `deploy-mode`: dos modos de despliegue\n",
    "\n",
    "    -   `client`: ejecuta el driver en el nodo local\n",
    "\n",
    "    -   `cluster`: ejecuta el driver en un nodo del cluster\n",
    "\n",
    "-   `class`: clase a ejecutar (Java o Scala)\n",
    "\n",
    "-   `name`: nombre de la aplicación (se muestra en el Spark web)\n",
    "\n",
    "-   `jars`: ficheros jar a añadir al classpath (Java o Scala)\n",
    "\n",
    "-   `py-files`: archivos a añadir al PYTHONPATH (`.py`,`.zip`,`.egg`)\n",
    "\n",
    "-   `files`: ficheros de datos para la aplicación\n",
    "\n",
    "-   `executor-memory`: memoria total de cada ejecutor\n",
    "\n",
    "-   `driver-memory`: memoria del proceso driver\n",
    "\n",
    "Para más opciones: `spark-submit --help`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages\n",
      "Requirement already satisfied: py4j==0.10.7 in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages (from pyspark)\n",
      "\u001b[33mYou are using pip version 9.0.1, however version 18.0 is available.\n",
      "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create apache spark context\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local\", appName=\"Mi app\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop apache spark context\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Instalación Manual\n",
    "\n",
    "Siguiendo estas dos guías podemos configurar de manera manual un pequeño cluster con un nodo maestro y dos esclavos, pero es un proceso muy largo y tedioso:\n",
    "- https://www.tutorialspoint.com/apache_spark/apache_spark_installation.htm\n",
    "- https://www.linode.com/docs/databases/hadoop/install-configure-run-spark-on-top-of-hadoop-yarn-cluster/\n",
    "\n",
    "![Spark docker meme](media/09.meme_docker_spark.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instalación Usando Docker\n",
    "\n",
    "Con base a lo comentado arriba instalar manualmente spark es un proceso largo y tedioso, por lo cual usaremos nuestro **super poder Docker**.\n",
    "\n",
    "![Spark Docker](media/08.spark_docker.jpg)\n",
    "\n",
    "Vamos a usar el repositorio oficial de la comunidad europea de big data.\n",
    "\n",
    "- https://github.com/big-data-europe/docker-spark\n",
    "\n",
    "Como prerequisito obviamente es necesario tener instalado docker, y adicionalmente docker-compose.\n",
    "\n",
    "### Docker\n",
    "\n",
    "#### Prerequisites\n",
    "\n",
    "Download && install **docker**\n",
    "- [For Mac](https://download.docker.com/mac/stable/Docker.dmg)\n",
    "- [For Windows](https://download.docker.com/win/stable/InstallDocker.msi)\n",
    "- [For Linux](https://docs.docker.com/engine/getstarted/step_one/#docker-for-linux)\n",
    "\n",
    "Download && install **docker-compose**\n",
    "- [Instructions](https://docs.docker.com/compose/install/)\n",
    "\n",
    "Download && install **docker-machine**\n",
    "- [Instructions](https://docs.docker.com/machine/install-machine/)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Correr cluster\n",
    "\n",
    "Pasos para levantar el cluster:\n",
    "\n",
    "* Clonar repositorio\n",
    "\n",
    "```sh\n",
    "$ git clone https://github.com/big-data-europe/docker-spark.git\n",
    "```\n",
    "\n",
    "* Entrar al repositorio\n",
    "\n",
    "```sh\n",
    "$ cd docker-spark\n",
    "```\n",
    "\n",
    "* Correr servicios\n",
    "\n",
    "```sh\n",
    "$ docker-compose up\n",
    "```\n",
    "\n",
    "En el archivo `docker-compose.yml` está la configuración del cluster con el nodo master y los esclavos.\n",
    "\n",
    "```yml\n",
    "version: '2'\n",
    "services:\n",
    "  spark-master:\n",
    "    image: bde2020/spark-master:2.3.1-hadoop2.7\n",
    "    container_name: spark-master\n",
    "    ports:\n",
    "      - \"8090:8080\"\n",
    "      - \"7077:7077\"\n",
    "    environment:\n",
    "      - INIT_DAEMON_STEP=setup_spark\n",
    "      - \"constraint:node==<yourmasternode>\"\n",
    "\n",
    "  spark-worker-1:\n",
    "    image: bde2020/spark-worker:2.3.1-hadoop2.7\n",
    "    container_name: spark-worker-1\n",
    "    depends_on:\n",
    "      - spark-master\n",
    "    ports:\n",
    "      - \"8091:8081\"\n",
    "    environment:\n",
    "      - \"SPARK_MASTER=spark://spark-master:7077\"\n",
    "      - \"constraint:node==<yourmasternode>\"\n",
    "\n",
    "  spark-worker-2:\n",
    "    image: bde2020/spark-worker:2.3.1-hadoop2.7\n",
    "    container_name: spark-worker-2\n",
    "    depends_on:\n",
    "      - spark-master\n",
    "    ports:\n",
    "      - \"8092:8081\"\n",
    "    environment:\n",
    "      - \"SPARK_MASTER=spark://spark-master:7077\"\n",
    "      - \"constraint:node==<yourworkernode>\"\n",
    "\n",
    "  spark-worker-3:\n",
    "    image: bde2020/spark-worker:2.3.1-hadoop2.7\n",
    "    container_name: spark-worker-3\n",
    "    depends_on:\n",
    "      - spark-master\n",
    "    ports:\n",
    "      - \"8093:8081\"\n",
    "    environment:\n",
    "      - \"SPARK_MASTER=spark://spark-master:7077\"\n",
    "      - \"constraint:node==<yourworkernode>\"\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Comprobar estado del cluster\n",
    "\n",
    "Usando el comando `docker ps` podemos ver todos los contenedores que estan corriendo en la máquina.\n",
    "\n",
    "![Dodkcer ps](media/11.docker_ps.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Luego se puede entrar a cada uno de los servicios usando el nombre del contenedor o el id:\n",
    "\n",
    "Comando:\n",
    "\n",
    "```sh\n",
    "$ docker exec -it <docker_id/docker_name> bash\n",
    "```\n",
    "\n",
    "Ejemplo:\n",
    "\n",
    "```sh\n",
    "$ docker exec -it spark-master bash\n",
    "```\n",
    "\n",
    "Comprobar spark:\n",
    "\n",
    "```sh\n",
    "$  ./spark/bin/spark-submit --help\n",
    "```\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark YARN\n",
    "\n",
    "![Yarn client mode](media/12.yarn_client_mod.png)\n",
    "![Yarn cluster mode](media/13.yarn_cluster_mode.png)\n",
    "\n",
    "\n",
    "Fuente: [Spark-on-YARN: Empower Spark Applications on Hadoop Cluster](https://www.slideshare.net/Hadoop_Summit/sparkonyarn-empower-spark-applications-on-hadoop-cluster)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parámetros de configuración\n",
    "\n",
    "Diversos parámetros ajustables en tiempo de ejecución\n",
    "\n",
    "-   En el script\n",
    "```python\n",
    "from pyspark import SparkConf,SparkContext\n",
    "conf = SparkConf()\n",
    "conf.set(\"spark.app.name\", \"Mi apli\")\n",
    "conf.set(\"spark.master\", \"local[2]\") # Cluster manager modo local con 2 hilos\n",
    "conf.set(\"spark.ui.port\", \"3600\")    # Puerto del interfaz web de Spark (por defecto: 4040)\n",
    "sc = SparkContext(conf=conf)\n",
    "```\n",
    "\n",
    "-   Mediante flags en el `spark-submit`\n",
    "\n",
    "```sh\n",
    "$ bin/spark-submit --master local[2] --name \"Mi apli\" \\  \n",
    "    --conf spark.ui.port=3600 mi-script.py\n",
    "```    \n",
    "    \n",
    "-   Mediante un fichero de propiedades\n",
    "\n",
    "```sh\n",
    "$ cat config.conf\n",
    "spark.master     local[2] \n",
    "spark.app.name   \"Mi apli\" \n",
    "spark.ui.port 3600\n",
    "$ bin/spark-submit --properties-file config.conf mi-script.py\n",
    "```\n",
    "\n",
    "Más info: <http://spark.apache.org/docs/latest/configuration.html#spark-properties>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejemplo\n",
    "\n",
    "Crear un archivo con el siguiente script comando:\n",
    "\n",
    "Script para crear archivo:\n",
    "\n",
    "```sh\n",
    "cat << EOF > /tmp/miscript.py\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from operator import add\n",
    "\n",
    "def main():\n",
    "    conf = SparkConf()\n",
    "    conf.set(\"spark.app.name\", \"Mi script Python\")\n",
    "\n",
    "    # Iniciamos el SparkContext\n",
    "    sc = SparkContext(conf=conf)\n",
    "    sc.setLogLevel(\"FATAL\")\n",
    "\n",
    "    rdd = sc.parallelize(range(100000)).cache()\n",
    "    \n",
    "    rdd2 = rdd.map(lambda x: (x, 2*x))\\\n",
    "              .map(lambda (x,y): (x-100, y**2))\\\n",
    "              .reduceByKey(lambda x,y: x+y)\\\n",
    "              .values()\n",
    "               \n",
    "    r = rdd2.reduce(add)\n",
    "    \n",
    "    print(\"Resultado final = {0}\".format(r))\n",
    "    \n",
    "    # Finalizamos el SparkContext\n",
    "    sc.stop()\n",
    "if __name__ == \"__main__\":\n",
    "    main()\n",
    "EOF\n",
    "```\n",
    "\n",
    "Correr script:\n",
    "\n",
    "```sh\n",
    "$ ./spark/bin/spark-submit --master local[8] /tmp/miscript.py\n",
    "```\n",
    "\n",
    "```sh\n",
    "$ ./spark-submit --master spark://965b174781c9:7077 --class org.apache.spark.examples.SparkPi /spark/examples/jars/spark-examples_2.11-2.2.0.jar 10\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}