Lab 2: Spark SQL - Student.ipynb

{
    "metadata": {
        "language_info": {
            "version": "2.7.11", 
            "file_extension": ".py", 
            "codemirror_mode": {
                "version": 2, 
                "name": "ipython"
            }, 
            "mimetype": "text/x-python", 
            "nbconvert_exporter": "python", 
            "name": "python", 
            "pygments_lexer": "ipython2"
        }, 
        "kernelspec": {
            "display_name": "Python 2 with Spark 2.0", 
            "language": "python", 
            "name": "python2-spark20"
        }
    }, 
    "nbformat": 4, 
    "cells": [
        {
            "metadata": {
                "collapsed": true
            }, 
            "cell_type": "markdown", 
            "source": "<img src='https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/sparkSQL4.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<img src='https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/sparkSQL2.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<img src='https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/sparkSQL3.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<img src='https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/sparkSQL1.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# SQL queries Dataframes, not RDDs"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "A data file on world banks will downloaded from GitHub after removing any previous data that may exist"
        }, 
        {
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "text": "--2017-06-07 21:31:48--  https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/world_bank.json.gz\nResolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.48.133\nConnecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.48.133|:443... connected.\nHTTP request sent, awaiting response... 200 OK\nLength: 446287 (436K) [application/octet-stream]\nSaving to: \u2018world_bank.json.gz\u2019\n\n100%[======================================>] 446,287     --.-K/s   in 0.02s   \n\n2017-06-07 21:31:48 (18.8 MB/s) - \u2018world_bank.json.gz\u2019 saved [446287/446287]\n\n", 
                    "name": "stdout"
                }
            ], 
            "cell_type": "code", 
            "execution_count": 2, 
            "source": "# In jupyter notebooks you can prefice commands with a ! to run shell commands\n# here we remove any files with the name of the file we are going to download\n# then download the file\n\n!rm world_bank.json.gz -f\n!wget https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/world_bank.json.gz"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Many other types are supported including text and Parquet\n\nHere we are creating a Dataframe, similar to an RDD, but with a schema and abstraction that allows\nfor SQL to be used."
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": 3, 
            "source": "#You can load json, text and other files using sqlContext\n#unlinke an RDD, this will attempt to create a schema around the data\n#self describing data works really well for this\n\nexample1_df = spark.read.json(\"./world_bank.json.gz\")"
        }, 
        {
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "text": "root\n |-- _id: struct (nullable = true)\n |    |-- $oid: string (nullable = true)\n |-- approvalfy: string (nullable = true)\n |-- board_approval_month: string (nullable = true)\n |-- boardapprovaldate: string (nullable = true)\n |-- borrower: string (nullable = true)\n |-- closingdate: string (nullable = true)\n |-- country_namecode: string (nullable = true)\n |-- countrycode: string (nullable = true)\n |-- countryname: string (nullable = true)\n |-- countryshortname: string (nullable = true)\n |-- docty: string (nullable = true)\n |-- envassesmentcategorycode: string (nullable = true)\n |-- grantamt: long (nullable = true)\n |-- ibrdcommamt: long (nullable = true)\n |-- id: string (nullable = true)\n |-- idacommamt: long (nullable = true)\n |-- impagency: string (nullable = true)\n |-- lendinginstr: string (nullable = true)\n |-- lendinginstrtype: string (nullable = true)\n |-- lendprojectcost: long (nullable = true)\n |-- majorsector_percent: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- Name: string (nullable = true)\n |    |    |-- Percent: long (nullable = true)\n |-- mjsector_namecode: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- code: string (nullable = true)\n |    |    |-- name: string (nullable = true)\n |-- mjtheme: array (nullable = true)\n |    |-- element: string (containsNull = true)\n |-- mjtheme_namecode: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- code: string (nullable = true)\n |    |    |-- name: string (nullable = true)\n |-- mjthemecode: string (nullable = true)\n |-- prodline: string (nullable = true)\n |-- prodlinetext: string (nullable = true)\n |-- productlinetype: string (nullable = true)\n |-- project_abstract: struct (nullable = true)\n |    |-- cdata: string (nullable = true)\n |-- project_name: string (nullable = true)\n |-- projectdocs: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- DocDate: string (nullable = true)\n |    |    |-- DocType: string (nullable = true)\n |    |    |-- DocTypeDesc: string (nullable = true)\n |    |    |-- DocURL: string (nullable = true)\n |    |    |-- EntityID: string (nullable = true)\n |-- projectfinancialtype: string (nullable = true)\n |-- projectstatusdisplay: string (nullable = true)\n |-- regionname: string (nullable = true)\n |-- sector: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- Name: string (nullable = true)\n |-- sector1: struct (nullable = true)\n |    |-- Name: string (nullable = true)\n |    |-- Percent: long (nullable = true)\n |-- sector2: struct (nullable = true)\n |    |-- Name: string (nullable = true)\n |    |-- Percent: long (nullable = true)\n |-- sector3: struct (nullable = true)\n |    |-- Name: string (nullable = true)\n |    |-- Percent: long (nullable = true)\n |-- sector4: struct (nullable = true)\n |    |-- Name: string (nullable = true)\n |    |-- Percent: long (nullable = true)\n |-- sector_namecode: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- code: string (nullable = true)\n |    |    |-- name: string (nullable = true)\n |-- sectorcode: string (nullable = true)\n |-- source: string (nullable = true)\n |-- status: string (nullable = true)\n |-- supplementprojectflg: string (nullable = true)\n |-- theme1: struct (nullable = true)\n |    |-- Name: string (nullable = true)\n |    |-- Percent: long (nullable = true)\n |-- theme_namecode: array (nullable = true)\n |    |-- element: struct (containsNull = true)\n |    |    |-- code: string (nullable = true)\n |    |    |-- name: string (nullable = true)\n |-- themecode: string (nullable = true)\n |-- totalamt: long (nullable = true)\n |-- totalcommamt: long (nullable = true)\n |-- url: string (nullable = true)\n\nNone\n", 
                    "name": "stdout"
                }
            ], 
            "cell_type": "code", 
            "execution_count": 4, 
            "source": "# Spark SQL has the ability to infer the schema of JSON data and understand the structure of the data\n#once we have created the Dataframe, we can print out the schema to see the shape of the data\n\nprint example1_df.printSchema()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Let's take a look at the first two rows of data\n\nThe example below enumerates our \"take\" command that pulls 2 items from the Dataframe\n<br>a simpiler option to see the data could also be:<br>\n\n##### copy and run the following code\n    for row in example1_df.take(2):\n        print row\n        print \"*\" * 20"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Now let's register a table which is a pointer to the Dataframe and allows data access via Spark SQL\n\n##### copy and run the following code\n    #Simply use the Dataframe Object to create the table:\n    example1_df.registerTempTable(\"world_bank\")"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### The returned object will be a dataframe\n##### copy and run the following code\n    temp_df =  spark.sql(\"select * from world_bank limit 2\")\n\n    print type(temp_df)\n    print \"*\" * 20\n    print temp_df"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": "\n"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### One nice feature of the notebooks and python is that we can show it in a table via Pandas\nspark.sql(\"select id, borrower from world_bank limit 2\").toPandas()"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Here is a simple group by example:\n\n#### Count the number of projects by each country, only list the top 10\n"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Simple Example of Adding a Schema (headers) to an RDD and using it as a dataframe"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### In the example below a simple RDD is created with Random Data in two columns and an ID column.\n\n#### copy and run the following code\n\n    import random\n\n    #first let's create a simple RDD\n\n    #create a Python list of lists for our example\n    data_e2 = []\n    for x in range(1,6):\n        random_int = int(random.random() * 10)\n        data_e2.append([x, random_int, random_int^2])\n\n    #create the RDD with the random list of lists\n    rdd_example2 = sc.parallelize(data_e2)\n    print rdd_example2.collect()\n"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### Now we can assign some header information\n\n#### copy and run the following code\n    from pyspark.sql.types import *\n\n    # The schema is encoded in a string.\n    schemaString = \"ID VAL1 VAL2\"\n\n    fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]\n    schema = StructType(fields)\n\n    # Apply the schema to the RDD.\n    schemaExample = sqlContext.createDataFrame(rdd_example2, schema)\n\n    # Register the DataFrame as a table.\n    schemaExample.registerTempTable(\"example2\")\n\n    # Pull the data\n    print schemaExample.collect()\n\n"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Print only first 2 rows of the rdd\n\n<br/>\n<div>\n<div class=\"panel panel-default\">\n    <div class=\"panel-heading\">\n      <h4 class=\"panel-title\">\n        <a data-toggle=\"collapse\" data-parent=\"#accordion-11\" href=\"#collapse1-11\">\n        Hint</a>\n      </h4>\n    </div>\n    <div id=\"collapse1-11\" class=\"panel-collapse collapse\">\n      <div class=\"panel-body\">Use .take(N) </div>\n    </div>\n  </div>\n </div>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### Now we can convert rdd_example3 to a Dataframe\n\n##### copy and run this code\n    from pyspark.sql import Row\n\n    rdd_example3 = rdd_example2.map(lambda x: Row(id=x[0], val1=x[1], val2=x[2]))\n    print rdd_example3.collect()\n    df_example3 = rdd_example3.toDF()"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Register this new data frame as a table \nregister as temp table, call it 'df_example3'"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Another powerful feature is the ability to create Functions and Use them in SQL Here is a simple example\n\nFirst we create a function in Python, then register it allowing for us to call it via SQL\n\n#### copy and run the following code\n    def simple_function(v):\n        return int(v * 10)\n\n    #test the function\n    print simple_function(3)"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### Now we can register the function for use in SQL\nspark.udf.register('simple_function',simple_function,pyspark.sql.types.IntegerType())\n"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### VAL1 and VAL2 look like strings, we can cast them as well\n\n    query = \"\"\"\n    select\n        ID,\n        VAL1,\n        VAL2,\n        simple_function(cast(VAL1 as int)) as s_VAL1,\n        simple_function(cast(VAL2 as int)) as s_VAL2\n    from\n     example2\n    \"\"\"\n    spark.sql(query).toPandas()"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Pandas Example\nPandas is a common abstraction for working with data in Python.\n\nWe can turn Pandas Dataframes into Spark Dataframes, the advantage of this \ncould be scale or allowing us to run SQL statements agains the data.\n\n### copy and run the following code\n    import pandas as pd\n    print pd"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### First, let's grab some UFO data to play with\n\n    !rm SIGHTINGS.csv -f\n    !wget https://www.quandl.com/api/v3/datasets/NUFORC/SIGHTINGS.csv"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Using the CSV file, we can create a Pandas Dataframe:\n    pandas_df = pd.read_csv(\"./SIGHTINGS.csv\")\n    pandas_df.head()\n    spark_df = spark.createDataFrame(pandas_df)"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### Look at the dataset, by registering as a temp table\n\n#### copy and paste this code\n\n    ufo_data = spark.createDataFrame(pandas_df)\n    ufo_data.registerTempTable('ufo_data')"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "# Visualizing the Data\n- Here are some simple ways to create charts using Pandas output\n- In order to display in the notebook we need to tell matplotlib to render inline\nat this point import the supporting libraries as well\n"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": 19, 
            "source": "%matplotlib inline \nimport matplotlib.pyplot as plt, numpy as np"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Pandas can call a function \"plot\" to create the charts.\nSince most charts are created from aggregates the record\nset should be small enough to store in Pandas\n\nWe can take our UFO data from before and create a \nPandas Dataframe from the Spark Dataframe\n\n    ufos_df = ufo_data.toPandas()"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "To plot we call the \"plot\" method and specify the type, x and y axis columns\nand optionally the size of the chart.\n\nMany more details can be found here:\nhttp://pandas.pydata.org/pandas-docs/stable/visualization.html\n    \n    \n#### copy and run this code\nufos_df.plot(kind='bar', x='Reports', y='Count', figsize=(12, 5))"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "### This doesn't look good, there are too many observations, can you check how many?\n"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<h2>Ideally we could just group by year, there are many ways we could solve that:</h2>\n\n1) parse the Reports column in SQL and output the year, then group on it\n2) create a simple Python function to parse the year and call it via sql\n3) as shown below: use map against the Dataframe and append a new column with \"year\"\n\nTge example below takes the existing data for each row and appends a new column \"year\" \nby taking the first for characters from the Reports column\n\nReports looks like this for example:\n2016-01-31\n\n##### copy and run this code\nufos_df = ufo_data.rdd.map(lambda x: Row(**dict(x.asDict(), year=int(x.Reports[0:4])))).toDF()"
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### Register the new Dataframe as a table \"ufo_withyear\""
        }, 
        {
            "metadata": {}, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "#### Now can you group by year, order by year and filter to the last 66 years?\nplot your results"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<div class=\"panel panel-default\">\n    <div class=\"panel-heading\">\n      <h4 class=\"panel-title\">\n        <a data-toggle=\"collapse\" data-parent=\"#accordion-11\" href=\"#collapse3-11\">\n        Answer</a>\n      </h4>\n    </div>\n    <div id=\"collapse3-11\" class=\"panel-collapse collapse\">\n      <div class=\"panel-body\">query = \"\"\"<br>\n            select <br>\n                sum(Count) as count, <br>\n                year <br>\n            from ufo_withyear<br>\n            where year > 1950<br>\n            group by year<br>\n            order by year<br>\n            \"\"\"<br>\n            pandas_ufos_withyears = sqlContext.sql(query).toPandas()<br>\n            pandas_ufos_withyears.plot(kind='bar', x='year', y='count', figsize=(12, 5))<br></div>\n    </div>\n  </div>"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "outputs": [], 
            "cell_type": "code", 
            "execution_count": null, 
            "source": ""
        }
    ], 
    "nbformat_minor": 1
}