Spark_DataFrame_API_Project/Spark_DataFrames_API.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init('/home/jubinsoni/spark-2.1.0-bin-hadoop2.7')\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('Basics').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.json('people.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "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": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(age=None, name='Michael')"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "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": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['age', 'name']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[summary: string, age: string, name: string]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+-------+\n",
      "|summary|               age|   name|\n",
      "+-------+------------------+-------+\n",
      "|  count|                 2|      3|\n",
      "|   mean|              24.5|   null|\n",
      "| stddev|7.7781745930520225|   null|\n",
      "|    min|                19|   Andy|\n",
      "|    max|                30|Michael|\n",
      "+-------+------------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import (StructField, StringType,\n",
    "                               IntegerType, StructType)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#age: name of variable, type of variable, True/False: if it can contain null\n",
    "data_schema = [StructField('age', IntegerType(), True),\n",
    "               StructField('name', StringType(), True)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[StructField(age,IntegerType,true), StructField(name,StringType,true)]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data_schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "final_struc = StructType(fields=data_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(age,IntegerType,true),StructField(name,StringType,true)))"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "final_struc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.json('people.json', schema=final_struc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: integer (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()\n",
    "#Now it has corrected the age as integer instead of long"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "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": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Column<b'age'>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df['age']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.column.Column"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df['age'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#If you want to select a dataframe column\n",
    "df.select('age')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "| age|\n",
      "+----+\n",
      "|null|\n",
      "|  30|\n",
      "|  19|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#If you want to view the results of dataframe column\n",
    "df.select('age').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name='Michael'), Row(age=30, name='Andy')]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#If you want to view selected rows of dataframe\n",
    "df.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(age=None, name='Michael')"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(2)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Michael'"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(2)[0]['name']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.types.Row"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df.head(2)[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "str"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df.head(2)[0]['name'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "##The reason there are so many datatypes and objects in Spark because of its ability\n",
    "#to read from a distributed source and map that out to compute"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[age: int, name: string]"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#selecting multiple columns\n",
    "df.select(['age', 'name'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "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": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "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": [
    "#Creating new columns\n",
    "df.withColumn('newAge', df['age']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "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": [
    "#Creating new columns\n",
    "df.withColumn('doubleAge', df['age']*2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#The above operations are not inplace so everytime you make\n",
    "#any changes you have to save them to a new variable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+\n",
      "|my_new_age|   name|\n",
      "+----------+-------+\n",
      "|      null|Michael|\n",
      "|        30|   Andy|\n",
      "|        19| Justin|\n",
      "+----------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Renaming a column\n",
    "df.withColumnRenamed('age', 'my_new_age').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working with Spark SQL queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#In order to work with SQL queries we need to register spark\n",
    "#As a temporary SQL view\n",
    "#We do CreateOrReplace incase we have already existing view it replaces\n",
    "\n",
    "df.createOrReplaceTempView('people')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Now you can pass direct SQL queries\n",
    "results = spark.sql(\"SELECT * FROM people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "results.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "age_results = spark.sql(\"SELECT * FROM people WHERE age=30\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+\n",
      "|age|name|\n",
      "+---+----+\n",
      "| 30|Andy|\n",
      "+---+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "age_results.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working with JSON, CSV and other formats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('ops').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#infer schema option is available with CSV\n",
    "\n",
    "df = spark.read.csv('appl_stock.csv', inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date: timestamp (nullable = true)\n",
      " |-- Open: double (nullable = true)\n",
      " |-- High: double (nullable = true)\n",
      " |-- Low: double (nullable = true)\n",
      " |-- Close: double (nullable = true)\n",
      " |-- Volume: integer (nullable = true)\n",
      " |-- Adj Close: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2010, 1, 4, 0, 0), Open=213.429998, High=214.499996, Low=212.38000099999996, Close=214.009998, Volume=123432400, Adj Close=27.727039),\n",
       " Row(Date=datetime.datetime(2010, 1, 5, 0, 0), Open=214.599998, High=215.589994, Low=213.249994, Close=214.379993, Volume=150476200, Adj Close=27.774976000000002),\n",
       " Row(Date=datetime.datetime(2010, 1, 6, 0, 0), Open=214.379993, High=215.23, Low=210.750004, Close=210.969995, Volume=138040000, Adj Close=27.333178000000004)]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 16 ms, total: 16 ms\n",
      "Wall time: 180 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Row(Date=datetime.datetime(2010, 1, 4, 0, 0), Open=213.429998, High=214.499996, Low=212.38000099999996, Close=214.009998, Volume=123432400, Adj Close=27.727039)"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df.head(1)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|                Date|              Open|              High|               Low|             Close|   Volume|         Adj Close|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|2010-01-04 00:00:...|        213.429998|        214.499996|212.38000099999996|        214.009998|123432400|         27.727039|\n",
      "|2010-01-05 00:00:...|        214.599998|        215.589994|        213.249994|        214.379993|150476200|27.774976000000002|\n",
      "|2010-01-06 00:00:...|        214.379993|            215.23|        210.750004|        210.969995|138040000|27.333178000000004|\n",
      "|2010-01-07 00:00:...|            211.75|        212.000006|        209.050005|            210.58|119282800|          27.28265|\n",
      "|2010-01-08 00:00:...|        210.299994|        212.000006|209.06000500000002|211.98000499999998|111902700|         27.464034|\n",
      "|2010-01-11 00:00:...|212.79999700000002|        213.000002|        208.450005|210.11000299999998|115557400|         27.221758|\n",
      "|2010-01-12 00:00:...|209.18999499999998|209.76999500000002|        206.419998|        207.720001|148614900|          26.91211|\n",
      "|2010-01-13 00:00:...|        207.870005|210.92999500000002|        204.099998|        210.650002|151473000|          27.29172|\n",
      "|2010-01-14 00:00:...|210.11000299999998|210.45999700000002|        209.020004|            209.43|108223500|         27.133657|\n",
      "|2010-01-15 00:00:...|210.92999500000002|211.59999700000003|        205.869999|            205.93|148516900|26.680197999999997|\n",
      "|2010-01-19 00:00:...|        208.330002|215.18999900000003|        207.240004|        215.039995|182501900|27.860484999999997|\n",
      "|2010-01-20 00:00:...|        214.910006|        215.549994|        209.500002|            211.73|153038200|         27.431644|\n",
      "|2010-01-21 00:00:...|        212.079994|213.30999599999998|        207.210003|        208.069996|152038600|         26.957455|\n",
      "|2010-01-22 00:00:...|206.78000600000001|        207.499996|            197.16|            197.75|220441900|         25.620401|\n",
      "|2010-01-25 00:00:...|202.51000200000001|        204.699999|        200.190002|        203.070002|266424900|26.309658000000002|\n",
      "|2010-01-26 00:00:...|205.95000100000001|        213.710005|        202.580004|        205.940001|466777500|         26.681494|\n",
      "|2010-01-27 00:00:...|        206.849995|            210.58|        199.530001|        207.880005|430642100|26.932840000000002|\n",
      "|2010-01-28 00:00:...|        204.930004|        205.500004|        198.699995|        199.289995|293375600|25.819922000000002|\n",
      "|2010-01-29 00:00:...|        201.079996|        202.199995|        190.250002|        192.060003|311488100|         24.883208|\n",
      "|2010-02-01 00:00:...|192.36999699999998|             196.0|191.29999899999999|        194.729998|187469100|         25.229131|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|              Open|\n",
      "+------------------+\n",
      "|        213.429998|\n",
      "|        214.599998|\n",
      "|        214.379993|\n",
      "|            211.75|\n",
      "|        210.299994|\n",
      "|212.79999700000002|\n",
      "|209.18999499999998|\n",
      "|        207.870005|\n",
      "|210.11000299999998|\n",
      "|210.92999500000002|\n",
      "|        208.330002|\n",
      "|        214.910006|\n",
      "|        212.079994|\n",
      "|206.78000600000001|\n",
      "|202.51000200000001|\n",
      "|205.95000100000001|\n",
      "|        206.849995|\n",
      "|        204.930004|\n",
      "|        201.079996|\n",
      "|192.36999699999998|\n",
      "+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#You can pass in conditions like \n",
    "df.filter('Close < 500').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|              Open|\n",
      "+------------------+\n",
      "|        213.429998|\n",
      "|        214.599998|\n",
      "|        214.379993|\n",
      "|            211.75|\n",
      "|        210.299994|\n",
      "|212.79999700000002|\n",
      "|209.18999499999998|\n",
      "|        207.870005|\n",
      "|210.11000299999998|\n",
      "|210.92999500000002|\n",
      "|        208.330002|\n",
      "|        214.910006|\n",
      "|        212.079994|\n",
      "|206.78000600000001|\n",
      "|202.51000200000001|\n",
      "|205.95000100000001|\n",
      "|        206.849995|\n",
      "|        204.930004|\n",
      "|        201.079996|\n",
      "|192.36999699999998|\n",
      "+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Opening price of every stock where Close price is less than 500\n",
    "df.filter('Close < 500').select('Open').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+------------------+\n",
      "|              Open|             Close|\n",
      "+------------------+------------------+\n",
      "|        213.429998|        214.009998|\n",
      "|        214.599998|        214.379993|\n",
      "|        214.379993|        210.969995|\n",
      "|            211.75|            210.58|\n",
      "|        210.299994|211.98000499999998|\n",
      "|212.79999700000002|210.11000299999998|\n",
      "|209.18999499999998|        207.720001|\n",
      "|        207.870005|        210.650002|\n",
      "|210.11000299999998|            209.43|\n",
      "|210.92999500000002|            205.93|\n",
      "|        208.330002|        215.039995|\n",
      "|        214.910006|            211.73|\n",
      "|        212.079994|        208.069996|\n",
      "|206.78000600000001|            197.75|\n",
      "|202.51000200000001|        203.070002|\n",
      "|205.95000100000001|        205.940001|\n",
      "|        206.849995|        207.880005|\n",
      "|        204.930004|        199.289995|\n",
      "|        201.079996|        192.060003|\n",
      "|192.36999699999998|        194.729998|\n",
      "+------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter('Close < 500').select(['Open', 'Close']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|                Date|              Open|              High|               Low|             Close|   Volume|         Adj Close|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|2010-01-04 00:00:...|        213.429998|        214.499996|212.38000099999996|        214.009998|123432400|         27.727039|\n",
      "|2010-01-05 00:00:...|        214.599998|        215.589994|        213.249994|        214.379993|150476200|27.774976000000002|\n",
      "|2010-01-06 00:00:...|        214.379993|            215.23|        210.750004|        210.969995|138040000|27.333178000000004|\n",
      "|2010-01-07 00:00:...|            211.75|        212.000006|        209.050005|            210.58|119282800|          27.28265|\n",
      "|2010-01-08 00:00:...|        210.299994|        212.000006|209.06000500000002|211.98000499999998|111902700|         27.464034|\n",
      "|2010-01-11 00:00:...|212.79999700000002|        213.000002|        208.450005|210.11000299999998|115557400|         27.221758|\n",
      "|2010-01-12 00:00:...|209.18999499999998|209.76999500000002|        206.419998|        207.720001|148614900|          26.91211|\n",
      "|2010-01-13 00:00:...|        207.870005|210.92999500000002|        204.099998|        210.650002|151473000|          27.29172|\n",
      "|2010-01-14 00:00:...|210.11000299999998|210.45999700000002|        209.020004|            209.43|108223500|         27.133657|\n",
      "|2010-01-15 00:00:...|210.92999500000002|211.59999700000003|        205.869999|            205.93|148516900|26.680197999999997|\n",
      "|2010-01-19 00:00:...|        208.330002|215.18999900000003|        207.240004|        215.039995|182501900|27.860484999999997|\n",
      "|2010-01-20 00:00:...|        214.910006|        215.549994|        209.500002|            211.73|153038200|         27.431644|\n",
      "|2010-01-21 00:00:...|        212.079994|213.30999599999998|        207.210003|        208.069996|152038600|         26.957455|\n",
      "|2010-01-22 00:00:...|206.78000600000001|        207.499996|            197.16|            197.75|220441900|         25.620401|\n",
      "|2010-01-25 00:00:...|202.51000200000001|        204.699999|        200.190002|        203.070002|266424900|26.309658000000002|\n",
      "|2010-01-26 00:00:...|205.95000100000001|        213.710005|        202.580004|        205.940001|466777500|         26.681494|\n",
      "|2010-01-27 00:00:...|        206.849995|            210.58|        199.530001|        207.880005|430642100|26.932840000000002|\n",
      "|2010-01-28 00:00:...|        204.930004|        205.500004|        198.699995|        199.289995|293375600|25.819922000000002|\n",
      "|2010-01-29 00:00:...|        201.079996|        202.199995|        190.250002|        192.060003|311488100|         24.883208|\n",
      "|2010-02-01 00:00:...|192.36999699999998|             196.0|191.29999899999999|        194.729998|187469100|         25.229131|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Close'] < 500).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|   Volume|\n",
      "+---------+\n",
      "|123432400|\n",
      "|150476200|\n",
      "|138040000|\n",
      "|119282800|\n",
      "|111902700|\n",
      "|115557400|\n",
      "|148614900|\n",
      "|151473000|\n",
      "|108223500|\n",
      "|148516900|\n",
      "|182501900|\n",
      "|153038200|\n",
      "|152038600|\n",
      "|220441900|\n",
      "|266424900|\n",
      "|466777500|\n",
      "|430642100|\n",
      "|293375600|\n",
      "|311488100|\n",
      "|187469100|\n",
      "+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Close'] < 500).select('Volume').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+----------+----------+----------+---------+------------------+\n",
      "|                Date|              Open|      High|       Low|     Close|   Volume|         Adj Close|\n",
      "+--------------------+------------------+----------+----------+----------+---------+------------------+\n",
      "|2010-01-22 00:00:...|206.78000600000001|207.499996|    197.16|    197.75|220441900|         25.620401|\n",
      "|2010-01-28 00:00:...|        204.930004|205.500004|198.699995|199.289995|293375600|25.819922000000002|\n",
      "|2010-01-29 00:00:...|        201.079996|202.199995|190.250002|192.060003|311488100|         24.883208|\n",
      "+--------------------+------------------+----------+----------+----------+---------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Multiple conditions in spark dataframe\n",
    "df.filter((df['Close'] < 200) & (df['Open'] > 200)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|                Date|              Open|              High|               Low|             Close|   Volume|         Adj Close|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "|2010-02-01 00:00:...|192.36999699999998|             196.0|191.29999899999999|        194.729998|187469100|         25.229131|\n",
      "|2010-02-02 00:00:...|        195.909998|        196.319994|193.37999299999998|        195.859997|174585600|25.375532999999997|\n",
      "|2010-02-03 00:00:...|        195.169994|        200.200003|        194.420004|        199.229994|153832000|25.812148999999998|\n",
      "|2010-02-04 00:00:...|        196.730003|        198.370001|        191.570005|        192.050003|189413000|         24.881912|\n",
      "|2010-02-05 00:00:...|192.63000300000002|             196.0|        190.850002|        195.460001|212576700|25.323710000000002|\n",
      "|2010-02-08 00:00:...|        195.690006|197.88000300000002|        193.999994|194.11999699999998|119567700|           25.1501|\n",
      "|2010-02-09 00:00:...|        196.419996|        197.499994|        194.749998|196.19000400000002|158221700|         25.418289|\n",
      "|2010-02-10 00:00:...|        195.889997|             196.6|            194.26|195.12000700000002| 92590400|          25.27966|\n",
      "|2010-02-11 00:00:...|        194.880001|        199.750006|194.05999599999998|        198.669994|137586400|         25.739595|\n",
      "|2010-02-23 00:00:...|        199.999998|        201.330002|        195.709993|        197.059998|143773700|         25.531005|\n",
      "|2014-06-09 00:00:...|         92.699997|         93.879997|             91.75|         93.699997| 75415000|         88.906324|\n",
      "|2014-06-10 00:00:...|         94.730003|         95.050003|             93.57|             94.25| 62777000|         89.428189|\n",
      "|2014-06-11 00:00:...|         94.129997|         94.760002|         93.470001|         93.860001| 45681000|         89.058142|\n",
      "|2014-06-12 00:00:...|         94.040001|         94.120003|         91.900002|         92.290001| 54749000|         87.568463|\n",
      "|2014-06-13 00:00:...|         92.199997|         92.440002|         90.879997|         91.279999| 54525000|         86.610132|\n",
      "|2014-06-16 00:00:...|         91.510002|             92.75|         91.449997|         92.199997| 35561000|         87.483064|\n",
      "|2014-06-17 00:00:...|         92.309998|         92.699997|         91.800003| 92.08000200000001| 29726000| 87.36920699999999|\n",
      "|2014-06-18 00:00:...|         92.269997|         92.290001|         91.349998|             92.18| 33514000|          87.46409|\n",
      "|2014-06-19 00:00:...|         92.290001|         92.300003|         91.339996|         91.860001| 35528000|         87.160461|\n",
      "|2014-06-20 00:00:...|         91.849998|         92.550003|         90.900002|         90.910004|100898000|         86.259066|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Not operator\n",
    "df.filter((df['Close'] < 200) & ~(df['Open'] > 200)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+----------+------+------+---------+---------+\n",
      "|                Date|              Open|      High|   Low| Close|   Volume|Adj Close|\n",
      "+--------------------+------------------+----------+------+------+---------+---------+\n",
      "|2010-01-22 00:00:...|206.78000600000001|207.499996|197.16|197.75|220441900|25.620401|\n",
      "+--------------------+------------------+----------+------+------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Low'] == 197.16).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2010, 1, 22, 0, 0), Open=206.78000600000001, High=207.499996, Low=197.16, Close=197.75, Volume=220441900, Adj Close=25.620401)]"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The above results look very messy and lot of times we want\n",
    "#to collect the data to work with\n",
    "df.filter(df['Low'] == 197.16).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "result = df.filter(df['Low'] == 197.16).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(Date=datetime.datetime(2010, 1, 22, 0, 0), Open=206.78000600000001, High=207.499996, Low=197.16, Close=197.75, Volume=220441900, Adj Close=25.620401)"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Adj Close': 25.620401,\n",
       " 'Close': 197.75,\n",
       " 'Date': datetime.datetime(2010, 1, 22, 0, 0),\n",
       " 'High': 207.499996,\n",
       " 'Low': 197.16,\n",
       " 'Open': 206.78000600000001,\n",
       " 'Volume': 220441900}"
      ]
     },
     "execution_count": 92,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "row = result[0]\n",
    "\n",
    "row.asDict()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "220441900"
      ]
     },
     "execution_count": 93,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "row.asDict()['Volume']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "220441900"
      ]
     },
     "execution_count": 94,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result[0]['Volume']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## GroupBy, Aggregate and Sorting Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('aggs').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv('sales_info.csv', inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "12"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Company: string (nullable = true)\n",
      " |-- Person: string (nullable = true)\n",
      " |-- Sales: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|Company| Person|Sales|\n",
      "+-------+-------+-----+\n",
      "|   GOOG|    Sam|200.0|\n",
      "|   GOOG|Charlie|120.0|\n",
      "|   GOOG|  Frank|340.0|\n",
      "|   MSFT|   Tina|600.0|\n",
      "|   MSFT|    Amy|124.0|\n",
      "|   MSFT|Vanessa|243.0|\n",
      "|     FB|   Carl|870.0|\n",
      "|     FB|  Sarah|350.0|\n",
      "|   APPL|   John|250.0|\n",
      "|   APPL|  Linda|130.0|\n",
      "|   APPL|   Mike|750.0|\n",
      "|   APPL|  Chris|350.0|\n",
      "+-------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|Company|count|\n",
      "+-------+-----+\n",
      "|   APPL|    4|\n",
      "|   GOOG|    3|\n",
      "|     FB|    2|\n",
      "|   MSFT|    3|\n",
      "+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Company').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+\n",
      "|Company|       avg(Sales)|\n",
      "+-------+-----------------+\n",
      "|   APPL|            370.0|\n",
      "|   GOOG|            220.0|\n",
      "|     FB|            610.0|\n",
      "|   MSFT|322.3333333333333|\n",
      "+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Company').mean().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+\n",
      "|Company|max(Sales)|\n",
      "+-------+----------+\n",
      "|   APPL|     750.0|\n",
      "|   GOOG|     340.0|\n",
      "|     FB|     870.0|\n",
      "|   MSFT|     600.0|\n",
      "+-------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Company').max().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+\n",
      "|Company|min(Sales)|\n",
      "+-------+----------+\n",
      "|   APPL|     130.0|\n",
      "|   GOOG|     120.0|\n",
      "|     FB|     350.0|\n",
      "|   MSFT|     124.0|\n",
      "+-------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Company').min().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|sum(Sales)|\n",
      "+----------+\n",
      "|    4327.0|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "## May be you dont need group by, you need something\n",
    "#like average sales per company\n",
    "\n",
    "df.agg({'Sales': 'sum'}).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|min(Sales)|\n",
      "+----------+\n",
      "|     120.0|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.agg({'Sales': 'max', 'Sales': 'min'}).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "group_data = df.groupBy('Company')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.group.GroupedData at 0x7fe465fefda0>"
      ]
     },
     "execution_count": 116,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "group_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+\n",
      "|Company|max(Sales)|\n",
      "+-------+----------+\n",
      "|   APPL|     750.0|\n",
      "|   GOOG|     340.0|\n",
      "|     FB|     870.0|\n",
      "|   MSFT|     600.0|\n",
      "+-------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#The agg method done using groupby\n",
    "group_data.agg({'Sales':'max'}).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import countDistinct, avg, stddev "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|count(DISTINCT Sales)|\n",
      "+---------------------+\n",
      "|                   11|\n",
      "+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Counts distinct sales value\n",
    "df.select(countDistinct('Sales')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|       avg(Sales)|\n",
      "+-----------------+\n",
      "|360.5833333333333|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(avg('Sales')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|    Average Sales|\n",
      "+-----------------+\n",
      "|360.5833333333333|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Alias - Give the column name appropriate name\n",
    "df.select(avg('Sales').alias('Average Sales')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|stddev_samp(Sales)|\n",
      "+------------------+\n",
      "|250.08742410799007|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(stddev('Sales')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Formatting the number\n",
    "from pyspark.sql.functions import format_number"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sales_std = df.select(stddev('Sales').alias('std'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|               std|\n",
      "+------------------+\n",
      "|250.08742410799007|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sales_std.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|format_number(std, 2)|\n",
      "+---------------------+\n",
      "|               250.09|\n",
      "+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#number of decimal places I want to show\n",
    "sales_std.select(format_number('std', 2)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|   std|\n",
      "+------+\n",
      "|250.09|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Again chaining name here\n",
    "sales_std.select(format_number('std', 2).alias('std')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|Company| Person|Sales|\n",
      "+-------+-------+-----+\n",
      "|   GOOG|Charlie|120.0|\n",
      "|   MSFT|    Amy|124.0|\n",
      "|   APPL|  Linda|130.0|\n",
      "|   GOOG|    Sam|200.0|\n",
      "|   MSFT|Vanessa|243.0|\n",
      "|   APPL|   John|250.0|\n",
      "|   GOOG|  Frank|340.0|\n",
      "|     FB|  Sarah|350.0|\n",
      "|   APPL|  Chris|350.0|\n",
      "|   MSFT|   Tina|600.0|\n",
      "|   APPL|   Mike|750.0|\n",
      "|     FB|   Carl|870.0|\n",
      "+-------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Ordering the columns, default is ascending order\n",
    "\n",
    "df.orderBy('Sales').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 136,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|Company| Person|Sales|\n",
      "+-------+-------+-----+\n",
      "|     FB|   Carl|870.0|\n",
      "|   APPL|   Mike|750.0|\n",
      "|   MSFT|   Tina|600.0|\n",
      "|     FB|  Sarah|350.0|\n",
      "|   APPL|  Chris|350.0|\n",
      "|   GOOG|  Frank|340.0|\n",
      "|   APPL|   John|250.0|\n",
      "|   MSFT|Vanessa|243.0|\n",
      "|   GOOG|    Sam|200.0|\n",
      "|   APPL|  Linda|130.0|\n",
      "|   MSFT|    Amy|124.0|\n",
      "|   GOOG|Charlie|120.0|\n",
      "+-------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Ordering in descending\n",
    "df.orderBy(df['Sales'].desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dealing with Missing Data in Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('miss').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv('ContainsNull.csv', header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John| null|\n",
      "|emp2| null| null|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#You can drop all the missing data\n",
    "df.na.drop().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John| null|\n",
      "|emp2| null| null|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 144,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John| null|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#When we specify thresh=2, it looks at the row if it contains more than 2 non-null values, it drops\n",
    "df.na.drop(thresh=2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 145,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#By default how is set to 'aby', meaning drop if there are any null values\n",
    "\n",
    "df.na.drop(how='any').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John| null|\n",
      "|emp2| null| null|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#We can change the how to 'all', if row has all null values only then drop\n",
    "\n",
    "df.na.drop(how='all').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#The subset method will only drop if Sales colum has null values\n",
    "df.na.drop(subset=['Sales']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Id: string (nullable = true)\n",
      " |-- Name: string (nullable = true)\n",
      " |-- Sales: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------+-----+\n",
      "|  Id|      Name|Sales|\n",
      "+----+----------+-----+\n",
      "|emp1|      John| null|\n",
      "|emp2|FILL VALUE| null|\n",
      "|emp3|FILL VALUE|345.0|\n",
      "|emp4|     Cindy|456.0|\n",
      "+----+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#We can also fill NULL values\n",
    "df.na.fill('FILL VALUE').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John|  0.0|\n",
      "|emp2| null|  0.0|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Spark is smart enough to recognize string datatype and only fill\n",
    "#in string datatype column, similar thing it does for numeric values\n",
    "df.na.fill(0).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+-----+\n",
      "|  Id|   Name|Sales|\n",
      "+----+-------+-----+\n",
      "|emp1|   John| null|\n",
      "|emp2|No Name| null|\n",
      "|emp3|No Name|345.0|\n",
      "|emp4|  Cindy|456.0|\n",
      "+----+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#You can specify the column name in the subset parameter\n",
    "#Its a best practice to declare subset always\n",
    "df.na.fill('No Name', subset=['Name']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 155,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "mean_val = df.select(mean(df['Sales'])).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(avg(Sales)=400.5)"
      ]
     },
     "execution_count": 157,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mean_val[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "400.5"
      ]
     },
     "execution_count": 158,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mean_val[0][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "mean_sales = mean_val[0][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 161,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John|400.5|\n",
      "|emp2| null|400.5|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.na.fill(mean_sales, ['Sales']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 167,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-----+\n",
      "|  Id| Name|Sales|\n",
      "+----+-----+-----+\n",
      "|emp1| John|400.5|\n",
      "|emp2| null|400.5|\n",
      "|emp3| null|345.0|\n",
      "|emp4|Cindy|456.0|\n",
      "+----+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Doing all this in one line\n",
    "df.na.fill(df.select(mean(df['Sales'])).collect()[0][0], ['Sales']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dates and Timestamps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 168,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('dates').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 170,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv('appl_stock.csv', header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 176,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1762"
      ]
     },
     "execution_count": 176,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 178,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2010, 1, 4, 0, 0), Open=213.429998, High=214.499996, Low=212.38000099999996, Close=214.009998, Volume=123432400, Adj Close=27.727039)]"
      ]
     },
     "execution_count": 178,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(1)\n",
    "#datetime object in Date column, this looks messy here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 179,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+\n",
      "|                Date|              Open|\n",
      "+--------------------+------------------+\n",
      "|2010-01-04 00:00:...|        213.429998|\n",
      "|2010-01-05 00:00:...|        214.599998|\n",
      "|2010-01-06 00:00:...|        214.379993|\n",
      "|2010-01-07 00:00:...|            211.75|\n",
      "|2010-01-08 00:00:...|        210.299994|\n",
      "|2010-01-11 00:00:...|212.79999700000002|\n",
      "|2010-01-12 00:00:...|209.18999499999998|\n",
      "|2010-01-13 00:00:...|        207.870005|\n",
      "|2010-01-14 00:00:...|210.11000299999998|\n",
      "|2010-01-15 00:00:...|210.92999500000002|\n",
      "|2010-01-19 00:00:...|        208.330002|\n",
      "|2010-01-20 00:00:...|        214.910006|\n",
      "|2010-01-21 00:00:...|        212.079994|\n",
      "|2010-01-22 00:00:...|206.78000600000001|\n",
      "|2010-01-25 00:00:...|202.51000200000001|\n",
      "|2010-01-26 00:00:...|205.95000100000001|\n",
      "|2010-01-27 00:00:...|        206.849995|\n",
      "|2010-01-28 00:00:...|        204.930004|\n",
      "|2010-01-29 00:00:...|        201.079996|\n",
      "|2010-02-01 00:00:...|192.36999699999998|\n",
      "+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(['Date', 'Open']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 180,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import (dayofmonth, hour,\n",
    "                                   dayofyear, month,\n",
    "                                   year, weekofyear,\n",
    "                                  format_number, date_format)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 182,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|dayofmonth(Date)|\n",
      "+----------------+\n",
      "|               4|\n",
      "|               5|\n",
      "|               6|\n",
      "|               7|\n",
      "|               8|\n",
      "|              11|\n",
      "|              12|\n",
      "|              13|\n",
      "|              14|\n",
      "|              15|\n",
      "|              19|\n",
      "|              20|\n",
      "|              21|\n",
      "|              22|\n",
      "|              25|\n",
      "|              26|\n",
      "|              27|\n",
      "|              28|\n",
      "|              29|\n",
      "|               1|\n",
      "+----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Whenever you are using any pyspark imported function,\n",
    "#you can directly use them in select on your column\n",
    "df.select(dayofmonth(df['Date'])).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 183,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|hour(Date)|\n",
      "+----------+\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "|         0|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(hour(df['Date'])).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 184,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|month(Date)|\n",
      "+-----------+\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          1|\n",
      "|          2|\n",
      "+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(month(df['Date'])).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 188,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|year(Date)|\n",
      "+----------+\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "|      2010|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "## MAy be you want to know average closing price pwe year\n",
    "\n",
    "df.select(year(df['Date'])).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 190,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+----+\n",
      "|                Date|              Open|              High|               Low|             Close|   Volume|         Adj Close|Year|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+----+\n",
      "|2010-01-04 00:00:...|        213.429998|        214.499996|212.38000099999996|        214.009998|123432400|         27.727039|2010|\n",
      "|2010-01-05 00:00:...|        214.599998|        215.589994|        213.249994|        214.379993|150476200|27.774976000000002|2010|\n",
      "|2010-01-06 00:00:...|        214.379993|            215.23|        210.750004|        210.969995|138040000|27.333178000000004|2010|\n",
      "|2010-01-07 00:00:...|            211.75|        212.000006|        209.050005|            210.58|119282800|          27.28265|2010|\n",
      "|2010-01-08 00:00:...|        210.299994|        212.000006|209.06000500000002|211.98000499999998|111902700|         27.464034|2010|\n",
      "|2010-01-11 00:00:...|212.79999700000002|        213.000002|        208.450005|210.11000299999998|115557400|         27.221758|2010|\n",
      "|2010-01-12 00:00:...|209.18999499999998|209.76999500000002|        206.419998|        207.720001|148614900|          26.91211|2010|\n",
      "|2010-01-13 00:00:...|        207.870005|210.92999500000002|        204.099998|        210.650002|151473000|          27.29172|2010|\n",
      "|2010-01-14 00:00:...|210.11000299999998|210.45999700000002|        209.020004|            209.43|108223500|         27.133657|2010|\n",
      "|2010-01-15 00:00:...|210.92999500000002|211.59999700000003|        205.869999|            205.93|148516900|26.680197999999997|2010|\n",
      "|2010-01-19 00:00:...|        208.330002|215.18999900000003|        207.240004|        215.039995|182501900|27.860484999999997|2010|\n",
      "|2010-01-20 00:00:...|        214.910006|        215.549994|        209.500002|            211.73|153038200|         27.431644|2010|\n",
      "|2010-01-21 00:00:...|        212.079994|213.30999599999998|        207.210003|        208.069996|152038600|         26.957455|2010|\n",
      "|2010-01-22 00:00:...|206.78000600000001|        207.499996|            197.16|            197.75|220441900|         25.620401|2010|\n",
      "|2010-01-25 00:00:...|202.51000200000001|        204.699999|        200.190002|        203.070002|266424900|26.309658000000002|2010|\n",
      "|2010-01-26 00:00:...|205.95000100000001|        213.710005|        202.580004|        205.940001|466777500|         26.681494|2010|\n",
      "|2010-01-27 00:00:...|        206.849995|            210.58|        199.530001|        207.880005|430642100|26.932840000000002|2010|\n",
      "|2010-01-28 00:00:...|        204.930004|        205.500004|        198.699995|        199.289995|293375600|25.819922000000002|2010|\n",
      "|2010-01-29 00:00:...|        201.079996|        202.199995|        190.250002|        192.060003|311488100|         24.883208|2010|\n",
      "|2010-02-01 00:00:...|192.36999699999998|             196.0|191.29999899999999|        194.729998|187469100|         25.229131|2010|\n",
      "+--------------------+------------------+------------------+------------------+------------------+---------+------------------+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('Year', year(df['Date'])).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 191,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "new_df = df.withColumn('Year', year(df['Date']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 192,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+------------------+------------------+------------------+--------------------+------------------+---------+\n",
      "|Year|         avg(Open)|         avg(High)|          avg(Low)|        avg(Close)|         avg(Volume)|    avg(Adj Close)|avg(Year)|\n",
      "+----+------------------+------------------+------------------+------------------+--------------------+------------------+---------+\n",
      "|2015|120.17575393253965|121.24452385714291| 118.8630954325397|120.03999980555547|  5.18378869047619E7|115.96740080555561|   2015.0|\n",
      "|2013| 473.1281355634922| 477.6389272301587|468.24710264682557| 472.6348802857143|          1.016087E8| 62.61798788492063|   2013.0|\n",
      "|2014| 295.1426195357143|297.56103184523823| 292.9949599801587| 295.4023416507935| 6.315273055555555E7| 87.63583323809523|   2014.0|\n",
      "|2012|     576.652720788| 581.8254008040001| 569.9211606079999| 576.0497195640002|       1.319642044E8| 74.81383696800002|   2012.0|\n",
      "|2016|104.50777772619044| 105.4271825436508|103.69027771825397|104.60400786904763|  3.84153623015873E7|103.15032854761901|   2016.0|\n",
      "|2010| 259.9576190992064|262.36880881349214|256.84761791269847| 259.8424600000002|1.4982631666666666E8|33.665072424603196|   2010.0|\n",
      "|2011|364.06142773412705| 367.4235704880951|360.29769878174613|364.00432532142867|1.2307474166666667E8| 47.16023692063492|   2011.0|\n",
      "+----+------------------+------------------+------------------+------------------+--------------------+------------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "new_df.groupBy('Year').mean().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 195,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+----+\n",
      "|        avg(Close)|Year|\n",
      "+------------------+----+\n",
      "|120.03999980555547|2015|\n",
      "| 472.6348802857143|2013|\n",
      "| 295.4023416507935|2014|\n",
      "| 576.0497195640002|2012|\n",
      "|104.60400786904763|2016|\n",
      "| 259.8424600000002|2010|\n",
      "|364.00432532142867|2011|\n",
      "+------------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "new_df.groupBy('Year').mean().select(['avg(Close)', 'Year']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 198,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "result = new_df.groupBy('Year').mean().select(['avg(Close)', 'Year'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 199,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+----+\n",
      "|        avg(Close)|Year|\n",
      "+------------------+----+\n",
      "|120.03999980555547|2015|\n",
      "| 472.6348802857143|2013|\n",
      "| 295.4023416507935|2014|\n",
      "| 576.0497195640002|2012|\n",
      "|104.60400786904763|2016|\n",
      "| 259.8424600000002|2010|\n",
      "|364.00432532142867|2011|\n",
      "+------------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 201,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "result = result.withColumnRenamed('avg(Close)', 'Average Closing Price')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 205,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------------------------------------+\n",
      "|Year|format_number(Average Closing Price, 2)|\n",
      "+----+---------------------------------------+\n",
      "|2015|                                 120.04|\n",
      "|2013|                                 472.63|\n",
      "|2014|                                 295.40|\n",
      "|2012|                                 576.05|\n",
      "|2016|                                 104.60|\n",
      "|2010|                                 259.84|\n",
      "|2011|                                 364.00|\n",
      "+----+---------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result.select(['Year', format_number('Average Closing Price', 2)]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 206,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------------------+\n",
      "|Year|Average Closing Price|\n",
      "+----+---------------------+\n",
      "|2015|               120.04|\n",
      "|2013|               472.63|\n",
      "|2014|               295.40|\n",
      "|2012|               576.05|\n",
      "|2016|               104.60|\n",
      "|2010|               259.84|\n",
      "|2011|               364.00|\n",
      "+----+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result.select(['Year', format_number('Average Closing Price', 2).alias('Average Closing Price')]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Thank you!"
   ]
  }
 ],
 "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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}