Lab-2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to Apache Spark: Querying data\n",
    "This notebook guides you through querying data with Apache Spark, including how to create and use DataFrames, run SQL queries, apply functions to the results of SQL queries, join data from different data sources, and visualize data in graphs.\n",
    "\n",
    "This notebook uses pySpark, the Python API for Spark. Some knowledge of Python is recommended. This notebook runs on Python 2 with Spark 1.6 and 2.0.\n",
    "\n",
    "If you are new to Apache Spark, see the first module in this series: __Introduction to Apache Spark: Basic Concepts__. (need link when published)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Table of contents\n",
    "\n",
    "1. [Prepare the environment and the data](#getstarted)<br>\n",
    "     1.1 [Enable SQL processing](#sqlprocessing)<br>\n",
    "     1.2 [Download the data](#download)<br>\n",
    "     1.3 [Create a DataFrame](#createdf)<br>\n",
    "     1.4 [Create a table](#createtab)<br>\n",
    "2. [Run SQL queries](#runsql)<br>\n",
    "    2.1 [Display query results with a pandas DataFrame](#pandas)<br>\n",
    "    2.2 [Run a group by query](#groupby)<br>\n",
    "    2.3 [Run a subselect query](#subselect)<br>\n",
    "    2.4 [Return nested JSON field values](#nested)<br>\n",
    "3. [Convert RDDs to DataFrames](#convertrdd)<br>\n",
    "    3.1 [Create a simple RDD](#simplerdd)<br>\n",
    "    3.2 [Apply a schema](#apply)<br>\n",
    "    3.3 [Create rows with named columns](#namedcol)<br>\n",
    "    3.4 [Join tables](#join)<br>\n",
    "4. [Create SQL functions](#sqlfuncs)<br>\n",
    "5. [Convert a pandas DataFrame to a Spark DataFrame](#sparkdf)<br>\n",
    "    5.1 [Get a new data set](#ufo)<br>\n",
    "    5.2 [Create a pandas DataFrame](#ufopandas)<br>\n",
    "    5.3 [Convert to a Spark DataFrame](#sparkufo)<br>\n",
    "    5.4 [Run an SQL statement](#runufo)<br>\n",
    "6. [Visualize data](#viz)<br>\n",
    "    6.1 [Create a chart](#vizchart)<br>\n",
    "    6.2 [Aggregate the data](#vizagg)<br>\n",
    "    6.3 [Create a better chart](#vizchart2)<br>\n",
    "7. [Summary and next steps](#nextsteps)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"getstarted\"></a>\n",
    "## 1. Prepare the environment and the data\n",
    "Before you can run SQL queries on data in an Apache Spark environment, you need to enable SQL processing and then move the data to the structured format of a DataFrame."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"sqlprocessing\"></a>\n",
    "### 1.1 Enable SQL processing\n",
    "The way you enable SQL processing with Spark 1.6 is to create an SQLContext. With Spark 2.0, the preferred method is the new SparkSession object, but the SQLContext object is still supported. \n",
    "\n",
    "Use the predefined Spark Context, `sc`, which contains the connection information for Spark to create an SQLContext:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"download\"></a>\n",
    "### 1.2 Download the data file\n",
    "\n",
    "You'll download a JSON file with data about world banks from GitHub.\n",
    "\n",
    "Remove any files with the same name as the file that you're going to download and then download the file from a URL:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2017-01-06 12:06:12--  https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/world_bank.json.gz\n",
      "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.48.133\n",
      "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.48.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 446287 (436K) [application/octet-stream]\n",
      "Saving to: ‘world_bank.json.gz’\n",
      "\n",
      "100%[======================================>] 446,287     --.-K/s   in 0.02s   \n",
      "\n",
      "2017-01-06 12:06:12 (17.2 MB/s) - ‘world_bank.json.gz’ saved [446287/446287]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!rm world_bank.json.gz -f\n",
    "!wget https://raw.githubusercontent.com/bradenrc/sparksql_pot/master/world_bank.json.gz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"createdf\"></a>\n",
    "### 1.3 Create a DataFrame \n",
    "\n",
    "Instead of creating an RDD to read the file, you'll create a Spark DataFrame. Unlike an RDD, a DataFrame creates a schema around the data, which supplies the necessary structure for SQL queries. A self-describing format like JSON is ideal for DataFrames, but many other file types are supported, including text (CSV) and Parquet.\n",
    "\n",
    "Create a DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "example1_df = sqlContext.read.json(\"./world_bank.json.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the schema to see how Spark SQL inferred the shape of the data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print example1_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now look at the first two rows of data.\n",
    "\n",
    "You can run the simple command `print example1_df.take(2)`, however, for readability, run the following command to include a row of asterisks in between the data rows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for row in example1_df.take(2):\n",
    "    print row\n",
    "    print \"*\" * 20"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"createtab\"></a>\n",
    "### 1.4 Create a table \n",
    "\n",
    "SQL statements must be run against a table. Create a table that's a pointer to the DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "example1_df.registerTempTable(\"world_bank\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"runsql\"></a>\n",
    "## 2. Run SQL queries\n",
    "\n",
    "You must define a new DataFrame for the results of the SQL query and put the SQL statement inside the `sqlContext.sql()` method.\n",
    "\n",
    "Run the following cell to select all columns from the table and print information about the resulting DataFrame and schema of the data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "temp_df =  sqlContext.sql(\"select * from world_bank\")\n",
    "\n",
    "print type(temp_df)\n",
    "print \"*\" * 20\n",
    "print temp_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first `print` command shows that the DataFrame is a Spark DataFrame. The last `print` command shows the column names and data types of the DataFrame."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"pandas\"></a>\n",
    "### 2.1 Display query results with a pandas DataFrame\n",
    "The `print` command doesn't show the data in a useful format. Instead of creating a Spark DataFrame, use the pandas open-source data analytics library to create a pandas DataFrame that shows the data in a table. \n",
    "\n",
    "Import the pandas library and use the `.toPandas()` method to show the query results:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "sqlContext.sql(\"select id, borrower from world_bank limit 2\").toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"groupby\"></a>\n",
    "### 2.2 Run a group by query\n",
    "\n",
    "You can make your SQL queries easier to read by using the `query` keyword and surrounding the SQL query with `\"\"\"` on separate lines. \n",
    "\n",
    "Calculate a count of projects by region:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "select\n",
    "    regionname ,\n",
    "    count(*) as project_count\n",
    "from world_bank\n",
    "group by regionname \n",
    "order by count(*) desc\n",
    "\"\"\"\n",
    "\n",
    "sqlContext.sql(query).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"subselect\"></a>\n",
    "### 2.3 Run a subselect query\n",
    "You can run subselect queries.\n",
    "\n",
    "Calculate a count of projects by region again, but this time using a subselect:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "\n",
    "select * from\n",
    "    (select\n",
    "        regionname ,\n",
    "        count(*) as project_count\n",
    "    from world_bank\n",
    "    group by regionname \n",
    "    order by count(*) desc) table_alias\n",
    "limit 2\n",
    "\"\"\"\n",
    "\n",
    "sqlContext.sql(query).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"nested\"></a>\n",
    "### 2.4 Return nested JSON field values\n",
    "With JSON data, you can select the values of nested fields with dot notation.\n",
    "\n",
    "Print the schema so that you can see that `sector.Name` is a nested field and then select its first two values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "example1_df.printSchema()\n",
    "\n",
    "sql = \"select sector.Name from world_bank limit 2\"\n",
    "sqlContext.sql(sql).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"convertrdd\"></a>\n",
    "## 3. Convert RDDs to DataFrames\n",
    "If you want to run SQL queries on an existing RDD, you must convert the RDD to a DataFrame. The main difference between and RDDs and DataFrames is whether the columns are named.\n",
    "\n",
    "You'll create an RDD and then convert it to a DataFrame in two different ways:\n",
    " - [Apply a schema](#apply)\n",
    " - [Create rows with named columns](#namedcol)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"simplerdd\"></a>\n",
    "### 3.1 Create a simple RDD\n",
    "You'll create a simple RDD with an ID column and two columns of random numbers.\n",
    "\n",
    "First create a Python list of lists:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import random\n",
    "\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])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now create the RDD:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd_example2 = sc.parallelize(data_e2)\n",
    "print rdd_example2.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"apply\"></a>\n",
    "### 3.2 Apply a schema\n",
    "You'll use the `StructField` method to create a schema object that's based on a string, apply the schema to the RDD to create a DataFrame, and then create a table to run SQL queries on.\n",
    "\n",
    "Define your schema columns as a string:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "\n",
    "schemaString = \"ID VAL1 VAL2\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Assign header information with the `StructField` method and create the schema with the `StructType` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]\n",
    "schema = StructType(fields)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Apply the schema to the RDD with the `createDataFrame` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schemaExample = sqlContext.createDataFrame(rdd_example2, schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Register the DataFrame as a table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schemaExample.registerTempTable(\"example2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "View the data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print schemaExample.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can reference the columns names in DataFrames:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for row in schemaExample.take(2):\n",
    "    print row.ID, row.VAL1, row.VAL2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run a simple SQL query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.sql(\"select * from example2\").toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"namedcol\"></a>\n",
    "### 3.3 Create rows with named columns\n",
    "You'll create an RDD with named columns and then convert it to a DataFrame and a table.\n",
    "\n",
    "Create a new RDD and specify the names of the columns with the `map` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "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",
    "\n",
    "print rdd_example3.collect()                                                            "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert `rdd_example3` to a DataFrame and register an associated table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_example3 = rdd_example3.toDF()\n",
    "df_example3.registerTempTable(\"example3\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run a simple SQL query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.sql(\"select * from example3\").toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"join\"></a>\n",
    "### 3.4 Join tables\n",
    "You can join tables.\n",
    "\n",
    "Join tables `example2` and `example3` on the ID column:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "select\n",
    "    *\n",
    "from\n",
    "    example2 e2\n",
    "inner join example3 e3 on\n",
    "    e2.ID = e3.id\n",
    "\"\"\"\n",
    "\n",
    "print sqlContext.sql(query).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, you can join DataFrames with a Python command instead of an SQL query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_example4 = df_example3.join(schemaExample, schemaExample[\"ID\"] == df_example3[\"id\"] )\n",
    "\n",
    "for row in df_example4.take(5):\n",
    "    print row"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"sqlfuncs\"></a>\n",
    "## 4. Create SQL functions \n",
    "You can create functions that run in SQL queries. \n",
    "\n",
    "First, create a Python function and test it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def simple_function(v):\n",
    "    return int(v * 10)\n",
    "\n",
    "#test the function\n",
    "print simple_function(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, register the function as an SQL function with the `registerFunction` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.registerFunction(\"simple_function\", simple_function)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now run the function in an SQL Statement:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "select\n",
    "    ID,\n",
    "    VAL1,\n",
    "    VAL2,\n",
    "    simple_function(VAL1) as s_VAL1,\n",
    "    simple_function(VAL2) as s_VAL2\n",
    "from\n",
    " example2\n",
    "\"\"\"\n",
    "sqlContext.sql(query).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The values in the VAL1 and VAL2 columns look like strings (10 characters instead of a number multiplied by 10). That's because string is the default data type for columns in Spark DataFrames.\n",
    "\n",
    "Cast the values in the VAL1 and VAL2 columns to integers: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "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",
    "sqlContext.sql(query).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That looks better!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "<a id=\"sparkdf\"></a>\n",
    "## 5. Convert a pandas DataFrame to a Spark DataFrame\n",
    "Although pandas DataFrames display data in a friendlier format, Spark DataFrames can be faster and more scalable.\n",
    "\n",
    "You'll get a new data set, create a pandas DataFrame for it, and then convert the pandas DataFrame to a Spark DataFrame."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"ufo\"></a>\n",
    "### 5.1 Get a new data set\n",
    "Get a data set about UFO sightings:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!rm SIGHTINGS.csv -f\n",
    "!wget https://www.quandl.com/api/v3/datasets/NUFORC/SIGHTINGS.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"ufopandas\"></a>\n",
    "### 5.2 Create a pandas DataFrame\n",
    "Create a pandas DataFrame of the data set with the `read_csv` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pandas_df = pd.read_csv(\"./SIGHTINGS.csv\")\n",
    "pandas_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"sparkufo\"></a>\n",
    "### 5.3 Convert to a Spark DataFrame\n",
    "Convert the pandas DataFrame to a Spark DataFrame with the `createDataFrame` method. Remember using the `createDataFrame` method to convert an RDD to a Spark DataFrame?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "spark_df = sqlContext.createDataFrame(pandas_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the first two rows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for row in spark_df.take(2):\n",
    "    print row\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Register the Spark DataFrame as a table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "spark_df.registerTempTable(\"ufo_sightings\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"runufo\"></a>\n",
    "### 5.4 Run an SQL statement\n",
    "Now run an SQL statement to print the first 10 rows of the table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print sqlContext.sql(\"select * from ufo_sightings limit 10\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"viz\"></a>\n",
    "## 6. Visualize data\n",
    "It's easy to create charts from pandas DataFrames. You'll use the matplotlib library to create graphs and the NumPy package for computing.\n",
    "\n",
    "Import the libraries and specify to show graphs inline:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%matplotlib inline \n",
    "import matplotlib.pyplot as plt, numpy as np"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert the Spark DataFrame with UFO data to a pandas DataFrame: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ufos_df = spark_df.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"vizchart\"></a>\n",
    "### 6.1 Create a chart\n",
    "\n",
    "To create a chart, you call the `plot()` method and specify the type of chart, the columns for the X and Y axes, and, optionally, the size of the chart. \n",
    "\n",
    "For more information about plotting pandas DataFrames, see [Visualization](http://pandas.pydata.org/pandas-docs/stable/visualization.html).\n",
    "\n",
    "Create a bar chart 12\" wide by 5\" high that shows the number of reports by date:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ufos_df.plot(kind='bar', x='Reports', y='Count', figsize=(12, 5))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This chart doesn't look good because there are too many observations. Check how many observations there are by querying the `ufo_sightings` table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print sqlContext.sql(\"select count(*) from ufo_sightings\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"vizagg\"></a>\n",
    "### 6.2 Aggregate the data\n",
    "\n",
    "To reduce the number of data points on the chart, you can aggregate the data by year. Here are a few of the ways that you can do that:\n",
    "\n",
    " - Run an SQL query the Reports column to output the year and then run a group by operation on the year.\n",
    " - Create a simple Python function to aggregate by year and then run the function in an SQL query.\n",
    " - Run the `map()` method on the Spark Dataframe to append a new column that contains the aggregated count by year. This is the method you'll use.\n",
    "\n",
    "Remember that the dates in the Reports column look like this: 2016-01-31. Therefore, to create the year column, you just need the first 4 characters from the Reports column.\n",
    "\n",
    "Add a year column to the DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#ufos_df = spark_df.map(lambda x: Row(**dict(x.asDict(), year=int(x.Reports[0:4])))).toDF()\n",
    "# new version\n",
    "ufos_df = spark_df.rdd.map(lambda x: Row(**dict(x.asDict(), year=int(x.Reports[0:4])))).toDF()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check to verify that you get the expected results:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print ufos_df.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Register the DataFrame as a table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ufos_df.registerTempTable(\"ufo_withyear\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"vizchart2\"></a>\n",
    "### 6.3 Create a better chart\n",
    "\n",
    "Now run an SQL query to group by year, order by year, and filter to the last 66 years. Then create a pandas DataFrame for the results and create a chart of the number of reports by year."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "select \n",
    "    sum(Count) as count, \n",
    "    year \n",
    "from ufo_withyear\n",
    "where year > 1950\n",
    "group by year\n",
    "order by year\n",
    "\"\"\"\n",
    "pandas_ufos_withyears = sqlContext.sql(query).toPandas()\n",
    "pandas_ufos_withyears.plot(kind='bar', x='year', y='count', figsize=(12, 5))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you have a chart that you can read. Look back at the original chart and notice that it wasn't ordered ascending by year."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "<a id=\"nextsteps\"></a>\n",
    "## 7. Summary and next steps\n",
    "You've learned how to create DataFrames, convert between DataFrame types, and convert from RDDs. You know how to run SQL queries and create SQL functions. And you can visualize the data in charts. \n",
    "\n",
    "Dig deeper:\n",
    " - [Apache Spark documentation](http://spark.apache.org/documentation.html)\n",
    " - [PySpark documentation](http://spark.apache.org/docs/latest/api/python/pyspark.html)\n",
    " - [pandas](http://pandas.pydata.org/pandas-docs/stable/index.html)\n",
    " - [matplotlib](http://matplotlib.org/)\n",
    " - [NumPy](http://www.numpy.org/)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Authors\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2 with Spark 1.6",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}