Optimizing for Data Skewness with Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Spark with Jupyter Notebook on MacOS (2.0.0 and higher)](https://medium.com/@roshinijohri/spark-with-jupyter-notebook-on-macos-2-0-0-and-higher-c61b971b5007)\n",
    "==========================================================================================\n",
    "\n",
    "#### Run in Terminal:\n",
    "$\\textrm{brew install apache-spark}$\n",
    "\n",
    "$\\textrm{brew info apache-spark}$\n",
    "\n",
    "$\\textrm{export SPARK_HOME='/usr/local/Cellar/apache-spark/2.4.5/libexec/'}$ -> Edit depending on version\n",
    "\n",
    "$\\textrm{pyspark}$"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\n",
      "      ____              __\n",
      "     / __/__  ___ _____/ /__\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 2.4.5\n",
      "      /_/\n",
      "\n",
      "Using Python version 3.7.4 (default, Aug 13 2019 15:17:50)\n",
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "exec(open(os.path.join(os.environ['SPARK_HOME'], 'python/pyspark/shell.py')).read())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql.session import SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "    .appName('spark test') \\\n",
    "    .getOrCreate() \\\n",
    "\n",
    "columns = ['id', 'dogs', 'cats']\n",
    "vals = [\n",
    "    (1, 2, 0),\n",
    "    (2, 0, 1)\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+----+\n",
      "| id|dogs|cats|\n",
      "+---+----+----+\n",
      "|  1|   2|   0|\n",
      "|  2|   0|   1|\n",
      "+---+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create DataFrame\n",
    "df = spark.createDataFrame(vals, columns)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark Broadcast"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.0.19:4041\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.5</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pyspark</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=pyspark>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Shut down the SparkContext\n",
    "sc.stop()\n",
    "\n",
    "# Main entry point for Spark functionality; A SparkContext represents the connection to a Spark cluster, \n",
    "#   and can be used to create L{RDD} and broadcast variables on that cluster\n",
    "# Note: master: Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4])\n",
    "#      appName: a name for your job, to display on the cluster web UI\n",
    "sc = SparkContext(master='local[*]', appName='pyspark')\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_dict = {'item1': 1, 'item2': 2, 'item3': 3, 'item4': 4} \n",
    "my_list = ['item1', 'item2', 'item3', 'item4']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Broadcast a read-only variable to the cluster, returning a L{Broadcast<pyspark.broadcast.Broadcast>} object for \n",
    "#  reading it in distributed functions. The variable will be sent to each cluster only once.\n",
    "my_dict_bc = sc.broadcast(my_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4]\n"
     ]
    }
   ],
   "source": [
    "def my_func(letter):\n",
    "    return my_dict_bc.value[letter] \n",
    "\n",
    "# Distribute a local Python collection to form an RDD\n",
    "# Using xrange is recommended if the input represents a range for performance\n",
    "my_list_rdd = sc.parallelize(my_list)\n",
    "\n",
    "# Return a new RDD by applying a function to each element of this RDD\n",
    "# Return a list that contains all of the elements in this RDD\n",
    "# Note: This method should only be used if the resulting array is expected to be small, as all the data \n",
    "#       is loaded into the driver’s memory\n",
    "result = my_list_rdd.map(lambda x: my_func(x)).collect()\n",
    "\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Optimizing for Data Skewness"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['_c0', 'Summons_Number', 'Plate_ID', 'Registration_State', 'Plate_Type', 'Issue_Date', 'Violation_Code', 'Vehicle_Body_Type', 'Vehicle_Make', 'Issuing_Agency', 'Street_Code1', 'Street_Code2', 'Street_Code3', 'Vehicle_Expiration_Date', 'Violation_Location', 'Violation_Precinct', 'Issuer_Precinct', 'Issuer_Code', 'Issuer_Command', 'Issuer_Squad', 'Violation_Time', 'Time_First_Observed', 'Violation_County', 'Violation_In_Front_Of_Or_Opposite', 'House_Number', 'Street_Name', 'Intersecting_Street', 'Date_First_Observed', 'Law_Section', 'Sub_Division', 'Violation_Legal_Code', 'Days_Parking_In_Effect____', 'From_Hours_In_Effect', 'To_Hours_In_Effect', 'Vehicle_Color', 'Unregistered_Vehicle?', 'Vehicle_Year', 'Meter_Number', 'Feet_From_Curb', 'Violation_Post_Code', 'Violation_Description', 'No_Standing_or_Stopping_Violation', 'Hydrant_Violation', 'Double_Parking_Violation', 'Latitude', 'Longitude', 'Community_Board', 'Community_Council_', 'Census_Tract', 'BIN', 'BBL', 'NTA', 'year', 'month']\n",
      "+----+-------+\n",
      "|year|  count|\n",
      "+----+-------+\n",
      "|2015|5986831|\n",
      "|2014|5821043|\n",
      "|2013|    592|\n",
      "|2000|    261|\n",
      "|2012|    189|\n",
      "|2011|    122|\n",
      "|2010|    110|\n",
      "|2001|     22|\n",
      "|2004|     22|\n",
      "|2005|     14|\n",
      "|2008|      6|\n",
      "|1988|      3|\n",
      "|2002|      3|\n",
      "|2009|      3|\n",
      "|2006|      3|\n",
      "|2003|      3|\n",
      "|2007|      3|\n",
      "|1986|      1|\n",
      "|1991|      1|\n",
      "|1985|      1|\n",
      "+----+-------+\n",
      "\n",
      "+-----+-------+\n",
      "|month|  count|\n",
      "+-----+-------+\n",
      "|    1|1392992|\n",
      "|    6|1276592|\n",
      "|    5|1040148|\n",
      "|    9|1029583|\n",
      "|    7| 969938|\n",
      "|   10| 966456|\n",
      "|    3| 965119|\n",
      "|    4| 951716|\n",
      "|    8| 911641|\n",
      "|   11| 798897|\n",
      "|   12| 774287|\n",
      "|    2| 731864|\n",
      "+-----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "def explore_dataframe():\n",
    "    \n",
    "    # Instantiate a Spark session \n",
    "    # The entry point to programming Spark with the Dataset and DataFrame API\n",
    "    # Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "    #       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "    spark = SparkSession.builder \\\n",
    "        .appName('Skewness Introduction') \\\n",
    "        .getOrCreate()\n",
    "\n",
    "    input_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Optimizing for Data Skewness with Spark/parking_violation.csv'\n",
    "\n",
    "    # Specifies the input data source format\n",
    "    # Note: source: string, name of the data source, e.g. 'json', 'parquet'\n",
    "    df = spark.read.format(source='csv').option('header', True).load(input_path)\n",
    "\n",
    "    # Investigate what columns you have\n",
    "    col_list = df.columns\n",
    "    print(col_list)\n",
    "\n",
    "    # groupby month and year to get count\n",
    "    year_df = df.groupby('year').count().sort('count', ascending=False)\n",
    "    month_df = df.groupby('month').count().sort('count', ascending=False)\n",
    "\n",
    "    year_df.show()\n",
    "    month_df.show()\n",
    "\n",
    "    # TODO write file partition by year, and study the executor in the spark UI\n",
    "    # TODO use repartition function\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    explore_dataframe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### partitionBy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "def partitionBy():\n",
    "\n",
    "    # Instantiate a Spark session \n",
    "    # The entry point to programming Spark with the Dataset and DataFrame API\n",
    "    # Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "    #       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "    spark = SparkSession.builder \\\n",
    "        .appName('Repartition Example') \\\n",
    "        .getOrCreate()\n",
    "    \n",
    "    input_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Optimizing for Data Skewness with Spark/parking_violation.csv'\n",
    "\n",
    "    # Specifies the input data source format\n",
    "    # Note: source: string, name of the data source, e.g. 'json', 'parquet'\n",
    "    df = spark.read.format(source='csv').option('header', True).load(input_path) \n",
    "    \n",
    "    # Partitions the output by the given columns on the file system\n",
    "    # Saves the content of the :class:`DataFrame` in CSV format at the specified path\n",
    "    # Note: path: the path in any Hadoop supported file system\n",
    "    #       mode: specifies the behavior of the save operation when data already exists\n",
    "    #             * ``append``: Append contents of this :class:`DataFrame` to existing data.\n",
    "    #             * ``overwrite``: Overwrite existing data.\n",
    "    #             * ``ignore``: Silently ignore this operation if data already exists.\n",
    "    #             * ``error`` or ``errorifexists`` (default case): Throw an exception if data already exists\n",
    "    out_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Optimizing for Data Skewness with Spark/parking_violation_partition_year.csv'\n",
    "    df.write.partitionBy('year').csv(path=out_path, mode='ignore')\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    partitionBy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Repartition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[id: bigint, dogs: bigint, cats: bigint]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns a new :class:`DataFrame` partitioned by the given partitioning expressions\n",
    "#   The resulting :class:`DataFrame` is hash partitioned\n",
    "# Note: numPartitions: can be an int to specify the target number of partitions or a Column\n",
    "df.repartition(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.repartition(6).rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "def repartition():\n",
    "\n",
    "    # Instantiate a Spark session \n",
    "    # The entry point to programming Spark with the Dataset and DataFrame API\n",
    "    # Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "    #       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "    spark = SparkSession.builder \\\n",
    "        .appName('Repartition Example') \\\n",
    "        .getOrCreate()\n",
    "    \n",
    "    input_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Optimizing for Data Skewness with Spark/parking_violation.csv'\n",
    "\n",
    "    # Specifies the input data source format\n",
    "    # Note: source: string, name of the data source, e.g. 'json', 'parquet'\n",
    "    df = spark.read.format(source='csv').option('header', True).load(input_path) \n",
    "    \n",
    "    # Returns a new :class:`DataFrame` partitioned by the given partitioning expressions\n",
    "    #   The resulting :class:`DataFrame` is hash partitioned\n",
    "    # Note: numPartitions: can be an int to specify the target number of partitions or a Column\n",
    "    # Saves the content of the :class:`DataFrame` in CSV format at the specified path\n",
    "    # Note: path: the path in any Hadoop supported file system\n",
    "    #       mode: specifies the behavior of the save operation when data already exists\n",
    "    #             * ``append``: Append contents of this :class:`DataFrame` to existing data.\n",
    "    #             * ``overwrite``: Overwrite existing data.\n",
    "    #             * ``ignore``: Silently ignore this operation if data already exists.\n",
    "    #             * ``error`` or ``errorifexists`` (default case): Throw an exception if data already exists\n",
    "    out_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Optimizing for Data Skewness with Spark/parking_violation_repartition6.csv'\n",
    "    df.repartition(6).write.csv(path=out_path, mode='ignore')\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    repartition()"
   ]
  },
  {
   "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}