DataFrame_Basics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark DataFrame Basics\n",
    "\n",
    "Spark DataFrames are the workhouse and main way of working with Spark and Python post Spark 2.0. DataFrames act as powerful versions of tables, with rows and columns, easily handling large datasets. The shift to DataFrames provides many advantages:\n",
    "* A much simpler syntax\n",
    "* Ability to use SQL directly in the dataframe\n",
    "* Operations are automatically distributed across RDDs\n",
    "    \n",
    "If you've used R or even the pandas library with Python you are probably already familiar with the concept of DataFrames. Spark DataFrame expand on a lot of these concepts, allowing you to transfer that knowledge easily by understanding the simple syntax of Spark DataFrames. Remember that the main advantage to using Spark DataFrames vs those other programs is that Spark can handle data across many RDDs, huge data sets that would never fit on a single computer. That comes at a slight cost of some \"peculiar\" syntax choices, but after this course you will feel very comfortable with all those topics!\n",
    "\n",
    "Let's get started!\n",
    "\n",
    "## Creating a DataFrame\n",
    "\n",
    "First we need to start a SparkSession:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then start the SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# May take a little while on a local computer\n",
    "spark = SparkSession.builder.appName(\"Basics\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "\n",
    "You will first need to get the data from a file (or connect to a large distributed file like HDFS, we'll talk about this later once we move to larger datasets on AWS EC2)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We'll discuss how to read other options later.\n",
    "# This dataset is from Spark's examples\n",
    "\n",
    "# Might be a little slow locally\n",
    "df = spark.read.json('people.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Showing the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Note how data is missing!\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: long (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['age', 'name']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[summary: string, age: string, name: string]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some data types make it easier to infer schema (like tabular formats such as csv which we will show later). \n",
    "\n",
    "However you often have to set the schema yourself if you aren't dealing with a .read method that doesn't have inferSchema() built-in.\n",
    "\n",
    "Spark has all the tools you need for this, it just requires a very specific structure:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField,StringType,IntegerType,StructType"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we need to create the list of Structure fields\n",
    "    * :param name: string, name of the field.\n",
    "    * :param dataType: :class:`DataType` of the field.\n",
    "    * :param nullable: boolean, whether the field can be null (None) or not."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_schema = [StructField(\"age\", IntegerType(), True),StructField(\"name\", StringType(), True)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "final_struc = StructType(fields=data_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.json('people.json', schema=final_struc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: integer (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Grabbing the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Column<b'age'>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df['age']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.column.Column"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df['age'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('age')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df.select('age'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "| age|\n",
      "+----+\n",
      "|null|\n",
      "|  30|\n",
      "|  19|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('age').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name='Michael'), Row(age=30, name='Andy')]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns list of Row objects\n",
    "df.head(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Multiple Columns:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int, name: string]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(['age','name'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(['age','name']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating new columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+------+\n",
      "| age|   name|newage|\n",
      "+----+-------+------+\n",
      "|null|Michael|  null|\n",
      "|  30|   Andy|    30|\n",
      "|  19| Justin|    19|\n",
      "+----+-------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Adding a new column with a simple copy\n",
    "df.withColumn('newage',df['age']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+\n",
      "|supernewage|   name|\n",
      "+-----------+-------+\n",
      "|       null|Michael|\n",
      "|         30|   Andy|\n",
      "|         19| Justin|\n",
      "+-----------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Simple Rename\n",
    "df.withColumnRenamed('age','supernewage').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "More complicated operations to create new columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---------+\n",
      "| age|   name|doubleage|\n",
      "+----+-------+---------+\n",
      "|null|Michael|     null|\n",
      "|  30|   Andy|       60|\n",
      "|  19| Justin|       38|\n",
      "+----+-------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('doubleage',df['age']*2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+-----------+\n",
      "| age|   name|add_one_age|\n",
      "+----+-------+-----------+\n",
      "|null|Michael|       null|\n",
      "|  30|   Andy|         31|\n",
      "|  19| Justin|         20|\n",
      "+----+-------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('add_one_age',df['age']+1).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+--------+\n",
      "| age|   name|half_age|\n",
      "+----+-------+--------+\n",
      "|null|Michael|    null|\n",
      "|  30|   Andy|    15.0|\n",
      "|  19| Justin|     9.5|\n",
      "+----+-------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('half_age',df['age']/2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int, name: string, half_age: double]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.withColumn('half_age',df['age']/2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll discuss much more complicated operations later on!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using SQL\n",
    "\n",
    "To use SQL queries directly with the dataframe, you will need to register it to a temporary view:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register the DataFrame as a SQL temporary view\n",
    "df.createOrReplaceTempView(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql_results = spark.sql(\"SELECT * FROM people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int, name: string]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sql_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_results.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+\n",
      "|age|name|\n",
      "+---+----+\n",
      "| 30|Andy|\n",
      "+---+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM people WHERE age=30\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We won't really be focusing on using the SQL syntax for this course in general, but keep in mind it is always there for you to get you out of bind quickly with your SQL skills!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alright that is all we need to know for now!"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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.0"
  },
  "nbpresent": {
   "slides": {},
   "themes": {
    "default": "0535adbc-b74f-46cc-9cd6-4eabe2477c8e",
    "theme": {
     "0535adbc-b74f-46cc-9cd6-4eabe2477c8e": {
      "backgrounds": {
       "backgroundColor": {
        "background-color": "backgroundColor",
        "id": "backgroundColor"
       }
      },
      "id": "0535adbc-b74f-46cc-9cd6-4eabe2477c8e",
      "palette": {
       "backgroundColor": {
        "id": "backgroundColor",
        "rgb": [
         43,
         43,
         43
        ]
       },
       "headingColor": {
        "id": "headingColor",
        "rgb": [
         238,
         238,
         238
        ]
       },
       "linkColor": {
        "id": "linkColor",
        "rgb": [
         19,
         218,
         236
        ]
       },
       "mainColor": {
        "id": "mainColor",
        "rgb": [
         238,
         238,
         238
        ]
       }
      },
      "rules": {
       "a": {
        "color": "linkColor"
       },
       "h1": {
        "color": "headingColor",
        "font-family": "Oswald",
        "font-size": 7
       },
       "h2": {
        "color": "headingColor",
        "font-family": "Oswald",
        "font-size": 5
       },
       "h3": {
        "color": "headingColor",
        "font-family": "Oswald",
        "font-size": 3.75
       },
       "h4": {
        "color": "headingColor",
        "font-family": "Oswald",
        "font-size": 3
       },
       "h5": {
        "color": "headingColor",
        "font-family": "Oswald"
       },
       "h6": {
        "color": "headingColor",
        "font-family": "Oswald"
       },
       "h7": {
        "color": "headingColor",
        "font-family": "Oswald"
       },
       "li": {
        "color": "mainColor",
        "font-family": "Lato",
        "font-size": 5
       },
       "p": {
        "color": "mainColor",
        "font-family": "Lato",
        "font-size": 5
       }
      },
      "text-base": {
       "color": "mainColor",
       "font-family": "Lato",
       "font-size": 5
      }
     },
     "cc59980f-cb69-400a-b63a-1fb85ca73c8a": {
      "backgrounds": {
       "dc7afa04-bf90-40b1-82a5-726e3cff5267": {
        "background-color": "31af15d2-7e15-44c5-ab5e-e04b16a89eff",
        "id": "dc7afa04-bf90-40b1-82a5-726e3cff5267"
       }
      },
      "id": "cc59980f-cb69-400a-b63a-1fb85ca73c8a",
      "palette": {
       "19cc588f-0593-49c9-9f4b-e4d7cc113b1c": {
        "id": "19cc588f-0593-49c9-9f4b-e4d7cc113b1c",
        "rgb": [
         252,
         252,
         252
        ]
       },
       "31af15d2-7e15-44c5-ab5e-e04b16a89eff": {
        "id": "31af15d2-7e15-44c5-ab5e-e04b16a89eff",
        "rgb": [
         68,
         68,
         68
        ]
       },
       "50f92c45-a630-455b-aec3-788680ec7410": {
        "id": "50f92c45-a630-455b-aec3-788680ec7410",
        "rgb": [
         197,
         226,
         245
        ]
       },
       "c5cc3653-2ee1-402a-aba2-7caae1da4f6c": {
        "id": "c5cc3653-2ee1-402a-aba2-7caae1da4f6c",
        "rgb": [
         43,
         126,
         184
        ]
       },
       "efa7f048-9acb-414c-8b04-a26811511a21": {
        "id": "efa7f048-9acb-414c-8b04-a26811511a21",
        "rgb": [
         25.118061674008803,
         73.60176211453744,
         107.4819383259912
        ]
       }
      },
      "rules": {
       "a": {
        "color": "19cc588f-0593-49c9-9f4b-e4d7cc113b1c"
       },
       "blockquote": {
        "color": "50f92c45-a630-455b-aec3-788680ec7410",
        "font-size": 3
       },
       "code": {
        "font-family": "Anonymous Pro"
       },
       "h1": {
        "color": "19cc588f-0593-49c9-9f4b-e4d7cc113b1c",
        "font-family": "Merriweather",
        "font-size": 8
       },
       "h2": {
        "color": "19cc588f-0593-49c9-9f4b-e4d7cc113b1c",
        "font-family": "Merriweather",
        "font-size": 6
       },
       "h3": {
        "color": "50f92c45-a630-455b-aec3-788680ec7410",
        "font-family": "Lato",
        "font-size": 5.5
       },
       "h4": {
        "color": "c5cc3653-2ee1-402a-aba2-7caae1da4f6c",
        "font-family": "Lato",
        "font-size": 5
       },
       "h5": {
        "font-family": "Lato"
       },
       "h6": {
        "font-family": "Lato"
       },
       "h7": {
        "font-family": "Lato"
       },
       "li": {
        "color": "50f92c45-a630-455b-aec3-788680ec7410",
        "font-size": 3.25
       },
       "pre": {
        "font-family": "Anonymous Pro",
        "font-size": 4
       }
      },
      "text-base": {
       "color": "19cc588f-0593-49c9-9f4b-e4d7cc113b1c",
       "font-family": "Lato",
       "font-size": 4
      }
     }
    }
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}