Standalone_Mode_Architecture.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark provides a simple standalone deploy mode. You can launch a standalone cluster either manually, \n",
    "by starting a master and workers by hand, or use provided launch scripts. \n",
    "It is also possible to run these daemons on a single machine for testing.\n",
    "\n",
    "\n",
    "The launch scripts do not currently support Windows. To run a Spark cluster on Windows, \n",
    "start the master and workers by hand.(http://spark.apache.org/docs/latest/spark-standalone.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Architecture"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Driver Program (Spark Context) => Master(Cluster Manager) = > Workers( Executors=> cores/slots)\n",
    "\n",
    "# Master hosts a cluster Manager on it and keep track of avaliable resources on worker nodes\n",
    "# Within driver Program, we have SparkContext to initialize the application, coordinates the running of processes on cluster\n",
    "## In client mode, driver program is on local machine, in cluster mode driver program is launched on worker machine\n",
    "## (i.e. in spark shell in terminal , when we submit the script while launching appllication, it is cluster mode)\n",
    "\n",
    "## Workers are like nodes, independent machines\n",
    "\n",
    "## As we interactive in jupyter notebook it is client mode."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Data/Architecture.PNG\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Through this exercise, I will configure standalone mode cluster setting , \n",
    "## start by powershell command, access the running cluster in jupyter notebook to do the analysis"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Settings"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "## My settings in spark-env.cmd file \n",
    "\n",
    "## We call one worker, as one machine or one node\n",
    "set SPARK_WORKER_INSTANCES=2 (Exexcutors)\n",
    "set SPARK_WORKER_CORES=4  # number of cores that one Worker can use (default: all cores, here it is 4)\n",
    "set SPARK_EXECUTOR_CORES=2 # We need to mention explicitly number of cores per executor, otherwise only one executor is launched with all cores\n",
    "set SPARK_WORKER_MEMORY=2g #total amount of memory that can be used on one machine (Worker Node) for running Spark programs.\n",
    "set SPARK_MASTER_PORT=7077\n",
    "set SPARK_MASTER_WEBUI_PORT=8080      \n",
    "set SPARK_WORKER_WEBUI_PORT=8081\n",
    "set SPARK_DRIVER_MEMORY=1g\n",
    "set SPARK_PUBLIC_DNS=\"localhost\"\n",
    "\n",
    "\n",
    "## Register each worker with  4 cores, 2.0 GiB RAM having 2 executors, each executor with 2 cores"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "## Power Shell commands to start the master and slaves on the windows\n",
    "\n",
    "##For first powershell terminal for starting master:\n",
    "cd \"C:\\Spark\\spark-3.0.1-bin-hadoop2.7\"\n",
    "bin\\spark-class org.apache.spark.deploy.master.Master --host \"127.0.0.1\"\n",
    "\n",
    "##For second powershell terminal for starting slave\n",
    "cd \"C:\\Spark\\spark-3.0.1-bin-hadoop2.7\"\n",
    "bin\\spark-class org.apache.spark.deploy.worker.Worker  spark://127.0.0.1:7077 --host \"127.0.0.1\"\n",
    "\n",
    "## We can start more terminals if we need to launch more workers or slaves\n",
    "\n",
    "## We can access master webui at \"127.0.0.1:8080\""
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "##Third powershell terminal launch of the application itself, if we want to access it through spark-shell in the terminal itself. Only need to run for running spark in shell\n",
    "spark-shell --master spark://127.0.0.1:7077\n",
    "\n",
    "## We can access application at \"127.0.0.1:4040\"  for debugging"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Accessing Cluster in Jupyter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "## After doing settings and starting the terminal, I am accessing it here\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark=SparkSession.builder.appName(\"spark-shell\").master(\"spark://127.0.0.1:7077\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.format(\"json\").load(\"Data\\sparkify_log_small.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sparkContext.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## First running above setting this is how my WEB UI look:\n",
    "\n",
    "2 executors per worker each with 2 cores and total memory (1g) (Worker with 2 gigs and 4 cores)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Data\\executors per worker.PNG\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Data\\Total_executors.PNG\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reflection\n",
    "####  We have three executors. By default, driver is also a executor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## If I look at the memory, \n",
    "## I have 1g memory divided into three parts equally among all executors. But when I check my master\n",
    "## Memory is allocated correctly. So Let us understand memory alocation\n",
    "\n",
    "## I have setting of driver memory , set SPARK_DRIVER_MEMORY=1g, I think it effects in some way. Let us figure it out\n",
    "## We have daemons: Master and Slaves\n",
    "##    SPARK_DAEMON_MEMORY\tMemory to allocate to the Spark master and worker daemons themselves (default: 1g)\n",
    "##    SPARK_WORKER_MEMORY\tTotal amount of memory to allow Spark applications to use on the machine, \n",
    "##   e.g. 1000m, 2g (default: total memory minus 1 GB); note that each application's individual memory \n",
    "## is configured using its spark.executor.memory property."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Driver will ask Master for resources, Master then allocates Workers to this application, \n",
    "# and Worker will start Executors, which are processes that run computations and store data for your application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## When I give SPARK_WORKER_MEMORY, i am keeping a portion of memory of machine for spark applications\n",
    "## SPARK_DAEMON_MEMORY = 1g (Memory to allocate to the Spark master and worker daemons themselves)\n",
    "## SPARK_WORKER_MEMORY = 2g\n",
    "## SPARK_EXECUTOR_MEMORY =THESE ARE APPLICATION SEPCIFIC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark=SparkSession.builder.appName(\"spark-shell2\").master(\"spark://127.0.0.1:7077\").config(\"spark.executor.memory\", \"1300m\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "##  \"spark.executor.memory\", \"1300m\" = JAVA HEAP\n",
    "reserved = 450\n",
    "spark memory = (1300-450)*0.6 = 510 => 513(on web ui)\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Driver Memory(System Memory) = 2G\n",
    "reservedMemory = 300\n",
    "minSystemMemory = (reservedMemory * 1.5) = 1.5 * 300 = 450\n",
    "usableMemory = systemMemory - reservedMemory = 2048 -450 = 1598M\n",
    "Spark memory = 0.6 *1598 = 958.8\n",
    "In conversion process bytes in all it become => 912 ( On WEB UI)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Driver Memory(System Memory) = 1G\n",
    "Spark memory = (1024-450)*0.6 = 344.4\n",
    "In conversion process, there will be slight variation =>366.3 (On web ui)"
   ]
  }
 ],
 "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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}