08 Deploying.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark training for data engineers\n",
    "## 08. Deploying\n",
    "\n",
    "### Goal\n",
    "Explain the logical steps to go from notebooks to a script to be deployed on a Spark cluster.\n",
    "\n",
    "### Highlights\n",
    "\n",
    "1. Combine all the important parts in one cell\n",
    "2. Restructure the code. First imports, next the functions and finally the main function\n",
    "3. Rename the code where needed\n",
    "4. Add docstrings and comments to the important steps of the main function\n",
    "5. Create a script from the combined cell\n",
    "6. Copy the script to the Spark cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Combine the code\n",
    "First step is to combine the code from the previous notebooks into one cell, so it can be executed at once."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "config = SparkConf().setMaster('local')\n",
    "spark = SparkContext.getOrCreate(conf=config)\n",
    "\n",
    "# Notebook 02\n",
    "csvrdd = spark.wholeTextFiles('file://///home/jovyan/*.csv')\n",
    "# Notebook 03\n",
    "csvrdd = csvrdd.flatMap(lambda x: x[1].split('\\n'))\n",
    "# Notebook 04\n",
    "from pyspark.sql import Row\n",
    "\n",
    "def processCSV(row):\n",
    "    # Split the row into a list\n",
    "    row = row.split(',')\n",
    "    # Return the four fields\n",
    "    return Row(row[0], row[1], row[2], row[3])\n",
    "\n",
    "csvrdd = csvrdd.map(lambda row: processCSV(row))\n",
    "csvrdd.collect()\n",
    "from pyspark.sql.types import StructField, StructType, StringType, IntegerType\n",
    "schema = StructType([\n",
    "    StructField(\"first_name\", StringType(), True),\n",
    "    StructField(\"last_name\", StringType(), True),\n",
    "    StructField(\"gender\", StringType(), True),\n",
    "    StructField(\"age\", StringType(), True)\n",
    "])\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(spark)\n",
    "csvdf = sqlContext.createDataFrame(csvrdd, schema=schema)\n",
    "# Notebook 05\n",
    "from pyspark.sql.functions import udf\n",
    "@udf('integer')\n",
    "def calc_name_length(name):\n",
    "    return len(name)\n",
    "csvdf = csvdf.withColumn('last_name_length', calc_name_length('last_name'))\n",
    "# Notebook 07\n",
    "csvdf.write.mode('overwrite').save(\"csvdf.json\", format=\"json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "part-00000-2247a872-6f05-4381-8253-f34f26be06e3-c000.json  _SUCCESS\n"
     ]
    }
   ],
   "source": [
    "ls *.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the script\n",
    "Create a script by restructuring the cell, moving the parts in a logical order."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql.types import StructField, StructType, StringType, IntegerType\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "@udf('integer')\n",
    "def calc_name_length(input_var):\n",
    "    \"\"\" Calculate the length of the input_var \"\"\"\n",
    "    return len(input_var)\n",
    "\n",
    "def processCSV(row):\n",
    "    \"\"\" Create a Spark Row from each CSV line \"\"\"\n",
    "    row = row.split(',')\n",
    "    return Row(row[0], row[1], row[2], row[3])\n",
    "\n",
    "config = SparkConf().setMaster('local')\n",
    "spark = SparkContext.getOrCreate(conf=config)\n",
    "sqlContext = SQLContext(spark)\n",
    "\n",
    "csvrdd = spark.wholeTextFiles('file://///home/jovyan/*.csv')\n",
    "csvrdd = csvrdd.flatMap(lambda x: x[1].split('\\n'))\n",
    "\n",
    "csvrdd = csvrdd.map(lambda row: processCSV(row))\n",
    "csvrdd.collect()\n",
    "schema = StructType([\n",
    "    StructField(\"first_name\", StringType(), True),\n",
    "    StructField(\"last_name\", StringType(), True),\n",
    "    StructField(\"gender\", StringType(), True),\n",
    "    StructField(\"age\", StringType(), True)\n",
    "])\n",
    "csvdf = sqlContext.createDataFrame(csvrdd, schema=schema)\n",
    "csvdf = csvdf.withColumn('last_name_length', calc_name_length('last_name'))\n",
    "csvdf.write.mode('overwrite').save(\"csvdf.json\", format=\"json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Clean up a little more:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import Row, SQLContext\n",
    "from pyspark.sql.types import StructField, StructType, StringType, IntegerType\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "@udf('integer')\n",
    "def calc_name_length(input_var):\n",
    "    \"\"\" Calculate the length of the input_var \"\"\"\n",
    "    return len(input_var)\n",
    "\n",
    "def processCSV(row):\n",
    "    \"\"\" Create a Spark Row from each CSV line \"\"\"\n",
    "    row = row.split(',')\n",
    "    return Row(row[0], row[1], row[2], row[3])\n",
    "\n",
    "def main():\n",
    "    \"\"\" Main function\"\"\"\n",
    "    \n",
    "    # Init\n",
    "    config = SparkConf().setMaster('local')\n",
    "    spark = SparkContext.getOrCreate(conf=config)\n",
    "    sqlContext = SQLContext(spark)\n",
    "    \n",
    "    # Process input\n",
    "    csvrdd = spark.wholeTextFiles('file://///home/jovyan/*.csv')\n",
    "    csvrdd = csvrdd.flatMap(lambda x: x[1].split('\\n'))\n",
    "    csvrdd = csvrdd.map(lambda row: processCSV(row))\n",
    "    \n",
    "    # Create dataframe\n",
    "    schema = StructType([\n",
    "        StructField(\"first_name\", StringType(), True),\n",
    "        StructField(\"last_name\", StringType(), True),\n",
    "        StructField(\"gender\", StringType(), True),\n",
    "        StructField(\"age\", StringType(), True)\n",
    "    ])\n",
    "    csvdf = sqlContext.createDataFrame(csvrdd, schema=schema)\n",
    "    \n",
    "    # Execute the custom functions\n",
    "    csvdf = csvdf.withColumn('last_name_length', calc_name_length('last_name'))\n",
    "\n",
    "    # Write the data\n",
    "    csvdf.write.mode('overwrite').save(\"csvdf.json\", format=\"json\")\n",
    "\n",
    "main()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the file from the above cell by using the `%%file` cell magic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting spark_script.py\n"
     ]
    }
   ],
   "source": [
    "%%file spark_script.py\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import Row, SQLContext\n",
    "from pyspark.sql.types import StructField, StructType, StringType, IntegerType\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "@udf('integer')\n",
    "def calc_length(input_var):\n",
    "    \"\"\" Calculate the length of the input_var \"\"\"\n",
    "    return len(input_var)\n",
    "\n",
    "def processCSV(row):\n",
    "    \"\"\" Create a Spark Row from each CSV line \"\"\"\n",
    "    row = row.split(',')\n",
    "    return Row(row[0], row[1], row[2], row[3])\n",
    "\n",
    "def main():\n",
    "    \"\"\" Main function\"\"\"\n",
    "    \n",
    "    # Init\n",
    "    config = SparkConf().setMaster('local')\n",
    "    spark = SparkContext.getOrCreate(conf=config)\n",
    "    sqlContext = SQLContext(spark)\n",
    "    \n",
    "    # Process input\n",
    "    csvrdd = spark.wholeTextFiles('file://///home/jovyan/*.csv')\n",
    "    csvrdd = csvrdd.flatMap(lambda x: x[1].split('\\n'))\n",
    "    csvrdd = csvrdd.map(lambda row: processCSV(row))\n",
    "    \n",
    "    # Create dataframe\n",
    "    schema = StructType([\n",
    "        StructField(\"first_name\", StringType(), True),\n",
    "        StructField(\"last_name\", StringType(), True),\n",
    "        StructField(\"gender\", StringType(), True),\n",
    "        StructField(\"age\", StringType(), True)\n",
    "    ])\n",
    "    csvdf = sqlContext.createDataFrame(csvrdd, schema=schema)\n",
    "    \n",
    "    # Execute the custom functions\n",
    "    csvdf = csvdf.withColumn('last_name_length', calc_length('last_name'))\n",
    "    \n",
    "    # Write the data\n",
    "    csvdf.write.mode('overwrite').save(\"csvdf.json\", format=\"json\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This script can be copied to the cluster and be executed with\n",
    "\n",
    "```\n",
    "$ spark-submit spark_script.py\n",
    "```"
   ]
  }
 ],
 "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}