DataFrame_operations_basics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark DataFrame basic operations\n",
    "### Dr. Tirthajyoti Sarkar, Fremont, CA, July 2020 (updated)\n",
    "In this notebook, we go through basic operations that can be performed with a Spark DataFrame object. We will use a .CSV file of stock prices to illustrate the various useful methods that are associated with a Spark DataFrame.\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### A `SparkSession` app, reading CSV, and the Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark1 = SparkSession.builder.appName('Ops').getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read the Apple stock CSV file. Note we have an option of inferring the schema for CSV. We also have the option to set `header` to `True`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark1.read.csv('Data/stock.csv',inferSchema=True,header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "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": "markdown",
   "metadata": {},
   "source": [
    "Show the DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-01-04 00:00:00|213.43| 214.5|212.38|214.01|123432400|    27.73|\n",
      "|2010-01-05 00:00:00| 214.6|215.59|213.25|214.38|150476200|    27.77|\n",
      "|2010-01-06 00:00:00|214.38|215.23|210.75|210.97|138040000|    27.33|\n",
      "|2010-01-07 00:00:00|211.75| 212.0|209.05|210.58|119282800|    27.28|\n",
      "|2010-01-08 00:00:00| 210.3| 212.0|209.06|211.98|111902700|    27.46|\n",
      "|2010-01-11 00:00:00| 212.8| 213.0|208.45|210.11|115557400|    27.22|\n",
      "|2010-01-12 00:00:00|209.19|209.77|206.42|207.72|148614900|    26.91|\n",
      "|2010-01-13 00:00:00|207.87|210.93| 204.1|210.65|151473000|    27.29|\n",
      "|2010-01-14 00:00:00|210.11|210.46|209.02|209.43|108223500|    27.13|\n",
      "|2010-01-15 00:00:00|210.93| 211.6|205.87|205.93|148516900|    26.68|\n",
      "|2010-01-19 00:00:00|208.33|215.19|207.24|215.04|182501900|    27.86|\n",
      "|2010-01-20 00:00:00|214.91|215.55| 209.5|211.73|153038200|    27.43|\n",
      "|2010-01-21 00:00:00|212.08|213.31|207.21|208.07|152038600|    26.96|\n",
      "|2010-01-22 00:00:00|206.78| 207.5|197.16|197.75|220441900|    25.62|\n",
      "|2010-01-25 00:00:00|202.51| 204.7|200.19|203.07|266424900|    26.31|\n",
      "|2010-01-26 00:00:00|205.95|213.71|202.58|205.94|466777500|    26.68|\n",
      "|2010-01-27 00:00:00|206.85|210.58|199.53|207.88|430642100|    26.93|\n",
      "|2010-01-28 00:00:00|204.93| 205.5| 198.7|199.29|293375600|    25.82|\n",
      "|2010-01-29 00:00:00|201.08| 202.2|190.25|192.06|311488100|    24.88|\n",
      "|2010-02-01 00:00:00|192.37| 196.0| 191.3|194.73|187469100|    25.23|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Show the columns\n",
    "\n",
    "The `columns` attribute returns a simple Python list of column names!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'Adj Close']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Spark DataFrames have separate `Column` and `Row` types \n",
    "Instead of generic Pandas series, Spark DataFrames have separate `Column` and `Row` types. This is ensure thet can handle special data files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.column.Column"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df['High'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.types.Row"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df.head(2)[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### The `select` method to select particular columns\n",
    "\n",
    "We can use this method to actually select the DataFrame columns and see them. Note that we still have to use the `show` method to actually output the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[High: double]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('High')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|  High|\n",
      "+------+\n",
      "| 214.5|\n",
      "|215.59|\n",
      "|215.23|\n",
      "| 212.0|\n",
      "| 212.0|\n",
      "| 213.0|\n",
      "|209.77|\n",
      "|210.93|\n",
      "|210.46|\n",
      "| 211.6|\n",
      "|215.19|\n",
      "|215.55|\n",
      "|213.31|\n",
      "| 207.5|\n",
      "| 204.7|\n",
      "|213.71|\n",
      "|210.58|\n",
      "| 205.5|\n",
      "| 202.2|\n",
      "| 196.0|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('High').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|  High| Close|\n",
      "+------+------+\n",
      "| 214.5|214.01|\n",
      "|215.59|214.38|\n",
      "|215.23|210.97|\n",
      "| 212.0|210.58|\n",
      "| 212.0|211.98|\n",
      "| 213.0|210.11|\n",
      "|209.77|207.72|\n",
      "|210.93|210.65|\n",
      "|210.46|209.43|\n",
      "| 211.6|205.93|\n",
      "|215.19|215.04|\n",
      "|215.55|211.73|\n",
      "|213.31|208.07|\n",
      "| 207.5|197.75|\n",
      "| 204.7|203.07|\n",
      "|213.71|205.94|\n",
      "|210.58|207.88|\n",
      "| 205.5|199.29|\n",
      "| 202.2|192.06|\n",
      "| 196.0|194.73|\n",
      "+------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(['High','Close']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### The `limit` method to take first few rows (without any collection)\n",
    "\n",
    "Applying `limit()` to the DataFrame will result in a new Dataframe. This is a transformation and does not perform collecting the data. Other similar methods like `take` and `head` result in an Array of Rows i.e. a Scala Collection Object like `scala.collection.immutable.Array` (which is transformed to Python list while using the PySpark API)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Date: timestamp, Open: double, High: double, Low: double, Close: double, Volume: int, Adj Close: double]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-01-04 00:00:00|213.43| 214.5|212.38|214.01|123432400|    27.73|\n",
      "|2010-01-05 00:00:00| 214.6|215.59|213.25|214.38|150476200|    27.77|\n",
      "|2010-01-06 00:00:00|214.38|215.23|210.75|210.97|138040000|    27.33|\n",
      "|2010-01-07 00:00:00|211.75| 212.0|209.05|210.58|119282800|    27.28|\n",
      "|2010-01-08 00:00:00| 210.3| 212.0|209.06|211.98|111902700|    27.46|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.limit(5).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### The `head()` and the `asDict()` methods"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2010, 1, 4, 0, 0), Open=213.43, High=214.5, Low=212.38, Close=214.01, Volume=123432400, Adj Close=27.73),\n",
       " Row(Date=datetime.datetime(2010, 1, 5, 0, 0), Open=214.6, High=215.59, Low=213.25, Close=214.38, Volume=150476200, Adj Close=27.77)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "dict1=df.head(2)[0].asDict()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Date': datetime.datetime(2010, 1, 4, 0, 0),\n",
       " 'Open': 213.43,\n",
       " 'High': 214.5,\n",
       " 'Low': 212.38,\n",
       " 'Close': 214.01,\n",
       " 'Volume': 123432400,\n",
       " 'Adj Close': 27.73}"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dict1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### The `count` method - number of rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1762"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Coutning rows with distinct values - `distinct` method\n",
    "\n",
    "To demonstrate this method, we read in another small CSV file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "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": [
    "df2 = spark1.read.csv('Data/sales_info.csv',inferSchema=True,header=True)\n",
    "df2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|Company|\n",
      "+-------+\n",
      "|   APPL|\n",
      "|   GOOG|\n",
      "|     FB|\n",
      "|   MSFT|\n",
      "+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.select('Company').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.select('Company').distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Random sampling using the `sample` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-03-19 00:00:00|224.79|225.24|221.23|222.25|139861400|    28.79|\n",
      "|2010-04-09 00:00:00|241.43|241.89|240.46|241.79| 83545700|    31.33|\n",
      "|2010-06-07 00:00:00|258.29|259.15|250.55|250.94|221735500|    32.51|\n",
      "|2011-07-19 00:00:00| 378.0|378.65|373.32|376.85|204786400|    48.82|\n",
      "|2012-07-24 00:00:00|607.38|609.68|598.51|600.92|141283100|    77.85|\n",
      "|2013-09-04 00:00:00|499.56|502.24|496.28|498.69| 86258200|    66.43|\n",
      "|2013-11-12 00:00:00|517.67|523.92| 517.0|520.01| 51069200|    69.68|\n",
      "|2013-11-19 00:00:00|519.03|523.38|517.97|519.55| 52234700|    69.62|\n",
      "|2014-02-06 00:00:00|510.06| 513.5|507.81|512.51| 64441300|    69.08|\n",
      "|2014-05-09 00:00:00|584.54|586.25|580.33|585.54| 72899400|    79.37|\n",
      "|2014-08-11 00:00:00| 95.27| 96.08| 94.84| 95.99| 36585000|    91.53|\n",
      "|2015-01-16 00:00:00|107.03|107.58| 105.2|105.99| 78513300|   101.51|\n",
      "|2016-11-11 00:00:00|107.12|108.87|106.55|108.43| 34143900|   107.96|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sample(withReplacement=False,\n",
    "          fraction = 0.005,\n",
    "          seed = 101).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Inserting new column using `withColumn` method\n",
    "\n",
    "We can insert a new column into the DataFrame using the `withColumn()` method. It is mostly used when we are doing some transformartion of the data using one or more columns. Following is a simple example where we are calculating the range of the stock price (difference between the `High` and `Low` columns).\n",
    "\n",
    "Also note how we are **chaining the methods** - a `withColumn`, followed by a `limit`, followed by a `select` to show only the columns impacted (in a certain order), and finally followed by a `show` to see the final result of only first five rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+\n",
      "|  High|   Low|             Range|\n",
      "+------+------+------------------+\n",
      "| 214.5|212.38|2.1200000000000045|\n",
      "|215.59|213.25|2.3400000000000034|\n",
      "|215.23|210.75|  4.47999999999999|\n",
      "| 212.0|209.05|2.9499999999999886|\n",
      "| 212.0|209.06|2.9399999999999977|\n",
      "+------+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('Range',df['High']-df['Low']).limit(5).select(['High','Low','Range']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is **not an _in-place_ operation** i.e. the original DataFrame remians unchanged."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Renaming a column using `withColumnRenamed` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = df.withColumn('Range',df['High']-df['Low']).limit(5).select(['High','Low','Range']) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+\n",
      "|  High|   Low|             Range|\n",
      "+------+------+------------------+\n",
      "| 214.5|212.38|2.1200000000000045|\n",
      "|215.59|213.25|2.3400000000000034|\n",
      "|215.23|210.75|  4.47999999999999|\n",
      "| 212.0|209.05|2.9499999999999886|\n",
      "| 212.0|209.06|2.9399999999999977|\n",
      "+------+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df3.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+\n",
      "|  High|   Low|        Min-to-Max|\n",
      "+------+------+------------------+\n",
      "| 214.5|212.38|2.1200000000000045|\n",
      "|215.59|213.25|2.3400000000000034|\n",
      "|215.23|210.75|  4.47999999999999|\n",
      "| 212.0|209.05|2.9499999999999886|\n",
      "| 212.0|209.06|2.9399999999999977|\n",
      "+------+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Just pass the old and new column names\n",
    "df3 = df3.withColumnRenamed('Range','Min-to-Max')\n",
    "df3.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Filtering Operations - using either `filter()` or `where()` method\n",
    "\n",
    "Filtering can be done with **SQL-like syntax** or **Pythonic** way. We show both examples."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Pass on a SQL syntax to the `filter()` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-01-04 00:00:00|213.43| 214.5|212.38|214.01|123432400|    27.73|\n",
      "|2010-01-05 00:00:00| 214.6|215.59|213.25|214.38|150476200|    27.77|\n",
      "|2010-01-06 00:00:00|214.38|215.23|210.75|210.97|138040000|    27.33|\n",
      "|2010-01-07 00:00:00|211.75| 212.0|209.05|210.58|119282800|    27.28|\n",
      "|2010-01-08 00:00:00| 210.3| 212.0|209.06|211.98|111902700|    27.46|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(\"Close < 500\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can chain methods to see only desired columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+\n",
      "|               Date|  Open| Close|\n",
      "+-------------------+------+------+\n",
      "|2012-02-15 00:00:00|514.26|497.67|\n",
      "|2013-09-05 00:00:00|500.25|495.27|\n",
      "|2013-09-10 00:00:00| 506.2|494.64|\n",
      "|2014-01-30 00:00:00|502.54|499.78|\n",
      "+-------------------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(\"Close < 500 AND Open > 500\").select(['Date','Open','Close']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+\n",
      "|               Date|  Open| Close|\n",
      "+-------------------+------+------+\n",
      "|2010-07-21 00:00:00|265.09|254.24|\n",
      "|2011-03-16 00:00:00| 342.0|330.01|\n",
      "|2011-08-04 00:00:00|389.41|377.37|\n",
      "|2011-09-29 00:00:00|401.92|390.57|\n",
      "|2011-11-10 00:00:00|397.03|385.22|\n",
      "+-------------------+------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.where(\"Open < 500 AND (Open-Close)> 10\").select(['Date','Open','Close']).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Now we use DataFrame syntax to achieve the same output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-01-04 00:00:00|213.43| 214.5|212.38|214.01|123432400|    27.73|\n",
      "|2010-01-05 00:00:00| 214.6|215.59|213.25|214.38|150476200|    27.77|\n",
      "|2010-01-06 00:00:00|214.38|215.23|210.75|210.97|138040000|    27.33|\n",
      "|2010-01-07 00:00:00|211.75| 212.0|209.05|210.58|119282800|    27.28|\n",
      "|2010-01-08 00:00:00| 210.3| 212.0|209.06|211.98|111902700|    27.46|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Close']<500).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**If we need to chain multiple conditions together, use `&` for AND and `|` for OR and clearly separate the conditions by putting them inside parantheses.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+\n",
      "|               Date|  Open| Close|\n",
      "+-------------------+------+------+\n",
      "|2012-02-15 00:00:00|514.26|497.67|\n",
      "|2013-09-05 00:00:00|500.25|495.27|\n",
      "|2013-09-10 00:00:00| 506.2|494.64|\n",
      "|2014-01-30 00:00:00|502.54|499.78|\n",
      "+-------------------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter((df['Close']<500) & (df['Open']>500)).select(['Date','Open','Close']).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+\n",
      "|               Date|  Open| Close|\n",
      "+-------------------+------+------+\n",
      "|2010-07-21 00:00:00|265.09|254.24|\n",
      "|2011-03-16 00:00:00| 342.0|330.01|\n",
      "|2011-08-04 00:00:00|389.41|377.37|\n",
      "|2011-09-29 00:00:00|401.92|390.57|\n",
      "|2011-11-10 00:00:00|397.03|385.22|\n",
      "+-------------------+------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter((df['Close']<500) & (df['Open']-df['Close']>10)).select(['Date','Open','Close']).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use `==` to compare with an exact value for comparison and `~` for NOT operator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+-----+------+------+---------+---------+\n",
      "|               Date|  Open| High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+-----+------+------+---------+---------+\n",
      "|2010-01-22 00:00:00|206.78|207.5|197.16|197.75|220441900|    25.62|\n",
      "+-------------------+------+-----+------+------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Low']==197.16).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-02-01 00:00:00|192.37| 196.0| 191.3|194.73|187469100|    25.23|\n",
      "|2010-02-02 00:00:00|195.91|196.32|193.38|195.86|174585600|    25.38|\n",
      "|2010-02-03 00:00:00|195.17| 200.2|194.42|199.23|153832000|    25.81|\n",
      "|2010-02-04 00:00:00|196.73|198.37|191.57|192.05|189413000|    24.88|\n",
      "|2010-02-05 00:00:00|192.63| 196.0|190.85|195.46|212576700|    25.32|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter((df['Open']<200) & ~(df['Close']>200)).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Use the `collect` method instead of `show`, to collect the actual data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "low_data = df.filter(df['Low']==197.16).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2010, 1, 22, 0, 0), Open=206.78, High=207.5, Low=197.16, Close=197.75, Volume=220441900, Adj Close=25.62)]"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "low_data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### It is still a list. So, grab the 0-index element as a Row object and convert it to a dictionary using `asDict` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "dt = low_data[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Date': datetime.datetime(2010, 1, 22, 0, 0),\n",
       " 'Open': 206.78,\n",
       " 'High': 207.5,\n",
       " 'Low': 197.16,\n",
       " 'Close': 197.75,\n",
       " 'Volume': 220441900,\n",
       " 'Adj Close': 25.62}"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dt.asDict()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, you can do whatever processing you want to do with the dictionary object!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Sorting DataFrame - `sort` and `orderBy` methods"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create a smaller DataFrame from the original one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|  High|   Low|\n",
      "+------+------+\n",
      "| 214.5|212.38|\n",
      "|215.59|213.25|\n",
      "|215.23|210.75|\n",
      "| 212.0|209.05|\n",
      "| 212.0|209.06|\n",
      "| 213.0|208.45|\n",
      "|209.77|206.42|\n",
      "|210.93| 204.1|\n",
      "|210.46|209.02|\n",
      "| 211.6|205.87|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df4 = df.select('High','Low').limit(10)\n",
    "df4.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|  High|   Low|\n",
      "+------+------+\n",
      "|209.77|206.42|\n",
      "|210.46|209.02|\n",
      "|210.93| 204.1|\n",
      "| 211.6|205.87|\n",
      "| 212.0|209.05|\n",
      "| 212.0|209.06|\n",
      "| 213.0|208.45|\n",
      "| 214.5|212.38|\n",
      "|215.23|210.75|\n",
      "|215.59|213.25|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df4.sort(\"High\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By default, it sorts by ascending order. You can change it by passing `ascending=False`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|  High|   Low|\n",
      "+------+------+\n",
      "|215.59|213.25|\n",
      "|215.23|210.75|\n",
      "| 214.5|212.38|\n",
      "| 213.0|208.45|\n",
      "| 212.0|209.05|\n",
      "| 212.0|209.06|\n",
      "| 211.6|205.87|\n",
      "|210.93| 204.1|\n",
      "|210.46|209.02|\n",
      "|209.77|206.42|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df4.sort(\"High\", ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also use the **`desc` function** imported from the `pyspark.sql.functions` module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|  High|   Low|\n",
      "+------+------+\n",
      "|215.59|213.25|\n",
      "|215.23|210.75|\n",
      "| 214.5|212.38|\n",
      "| 213.0|208.45|\n",
      "| 212.0|209.05|\n",
      "| 212.0|209.06|\n",
      "| 211.6|205.87|\n",
      "|210.93| 204.1|\n",
      "|210.46|209.02|\n",
      "|209.77|206.42|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import desc\n",
    "df4.sort(desc(\"High\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can pass a list to the `ascending` parameter to create custom sorting order on various columns. In the following example, we are sorting in descending order for the `Company` column, while doing an ascending sort on the `Sales` column. We are using the `orderBy` method for this one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "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": [
    "df2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|Company| Person|Sales|\n",
      "+-------+-------+-----+\n",
      "|   MSFT|    Amy|124.0|\n",
      "|   MSFT|Vanessa|243.0|\n",
      "|   MSFT|   Tina|600.0|\n",
      "|   GOOG|Charlie|120.0|\n",
      "|   GOOG|    Sam|200.0|\n",
      "|   GOOG|  Frank|340.0|\n",
      "|     FB|  Sarah|350.0|\n",
      "|     FB|   Carl|870.0|\n",
      "|   APPL|  Linda|130.0|\n",
      "|   APPL|   John|250.0|\n",
      "|   APPL|  Chris|350.0|\n",
      "|   APPL|   Mike|750.0|\n",
      "+-------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.orderBy([\"Company\",\"Sales\"],ascending=[0,1]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "### Using functions from the `pyspark.sql.functions` module\n",
    "\n",
    "We can utilize various utility functions from the `pyspark.sql.functions` module to do transformation on the data and extract insights, look for patterns, build statistical models, etc.\n",
    "\n",
    "We will just show a couple of examples here, but you are encouraged to explore others. Here is the detailed documentation link: \n",
    "\n",
    "http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import corr, dayofweek, month"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Suppose, we want to check ***if there is any correlation between the trade volumes and the range (Max - Min) i.e. intra-day volatility of the stock***.\n",
    "\n",
    "Here is how we can do that computation. Note the use of two new methods - `agg` which is nothing but an aggregration over the entire DataFrame without forming any specific groups and the `alias` method which is just a way to specify a descriptive name for the column resulting from the application of the `corr` function. We need the `agg` here because we want to aggregrate and throw all the data of the DataFrame at the `corr` function to calculate the [Pearson correlation coefficient](https://en.wikipedia.org/wiki/Pearson_correlation_coefficient) between the `Volume` and `Range` data columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Pearson-Corr=0.6980419831842759)]"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.withColumn('Range',df['High']-df['Low']).agg(corr(\"Volume\",\"Range\").alias('Pearson-Corr')).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So, we do have a fairly strong positive correlation between the `Volume` and `Range` data.\n",
    "\n",
    "Here is another fun observation! If you stare at the code above a bit longer, you will notice that **we did not produce a standalone `Range` column at all**. We computed it in an ephemeral manner - just enough to calculate the correlation coefficient. The original DataFrame remains unchanged while we got our answer about the positive correlation!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|               Date|  Open|  High|   Low| Close|   Volume|Adj Close|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "|2010-01-04 00:00:00|213.43| 214.5|212.38|214.01|123432400|    27.73|\n",
      "|2010-01-05 00:00:00| 214.6|215.59|213.25|214.38|150476200|    27.77|\n",
      "|2010-01-06 00:00:00|214.38|215.23|210.75|210.97|138040000|    27.33|\n",
      "|2010-01-07 00:00:00|211.75| 212.0|209.05|210.58|119282800|    27.28|\n",
      "|2010-01-08 00:00:00| 210.3| 212.0|209.06|211.98|111902700|    27.46|\n",
      "+-------------------+------+------+------+------+---------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we apply the `dayofweek` and `month` functions on the `Date` column to produce new columns `Day-of-Week` and `Month`. Note how we chained two `withColumn` methods together for this transformation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+------+------+\n",
      "|Day-of-Week|Month|  Open| Close|\n",
      "+-----------+-----+------+------+\n",
      "|          2|    1|213.43|214.01|\n",
      "|          3|    1| 214.6|214.38|\n",
      "|          4|    1|214.38|210.97|\n",
      "|          5|    1|211.75|210.58|\n",
      "|          6|    1| 210.3|211.98|\n",
      "+-----------+-----+------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df5=df.withColumn('Day-of-Week',dayofweek('Date')).\\\n",
    "withColumn('Month',month('Date')).\\\n",
    "select(['Day-of-Week','Month','Open','Close'])\n",
    "\n",
    "df5.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we have not shown the `groupby()` and aggregration (like average or `mean`) methods yet but if you have a fair understanding of what a `groupby` (followed by an aggregration like `mean`) does, then you can appreciate the following example where we look at the average `Volume` grouped by `Day-of-Week` and `Month` and try to gauge ***which day of the week or month have the highest and lowest average trade volumes***.\n",
    "\n",
    "The example below is certainly a complex one but you can think this one as an illustration of using some of the most important DataFrame methods we have examined in this Notebook (and some future ones we will look at), in a combined fashion.\n",
    "\n",
    "- `withColumn`\n",
    "- `select`\n",
    "- `orderBy`\n",
    "- `groupBy`\n",
    "- `mean` (aggregration)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+--------------------+\n",
      "|Day-of-Week|Month|         avg(Volume)|\n",
      "+-----------+-----+--------------------+\n",
      "|          2|    1|1.0967633636363636E8|\n",
      "|          2|    2| 8.957742272727273E7|\n",
      "|          2|    3| 9.421765161290322E7|\n",
      "|          2|    4|1.0364189333333333E8|\n",
      "|          2|    5| 9.646600416666667E7|\n",
      "|          2|    6| 8.898051666666667E7|\n",
      "|          2|    7| 8.163605555555555E7|\n",
      "|          2|    8|      8.5287746875E7|\n",
      "|          2|    9|  9.82580956521739E7|\n",
      "|          2|   10| 8.972772333333333E7|\n",
      "|          2|   11|          7.374762E7|\n",
      "|          2|   12| 7.757132068965517E7|\n",
      "|          3|    1|1.3081937586206897E8|\n",
      "|          3|    2|1.0092741428571428E8|\n",
      "|          3|    3|      1.0023139375E8|\n",
      "|          3|    4|1.0715501333333333E8|\n",
      "|          3|    5|1.0042301290322581E8|\n",
      "|          3|    6| 8.865190333333333E7|\n",
      "|          3|    7| 9.382107096774194E7|\n",
      "|          3|    8| 9.748764193548387E7|\n",
      "|          3|    9|          9.932478E7|\n",
      "|          3|   10|1.1340706896551724E8|\n",
      "|          3|   11|           7.94779E7|\n",
      "|          3|   12|6.3774113333333336E7|\n",
      "|          4|    1|1.2374592413793103E8|\n",
      "|          4|    2|1.0764395862068966E8|\n",
      "|          4|    3|1.0374595806451613E8|\n",
      "|          4|    4|1.1222539666666667E8|\n",
      "|          4|    5|1.0136810333333333E8|\n",
      "|          4|    6| 8.363113225806452E7|\n",
      "|          4|    7|          9.693284E7|\n",
      "|          4|    8| 8.988111935483871E7|\n",
      "|          4|    9|1.0007247333333333E8|\n",
      "|          4|   10| 9.734730967741935E7|\n",
      "|          4|   11| 7.615277666666667E7|\n",
      "|          4|   12|          7.015238E7|\n",
      "|          5|    1|1.0383316666666667E8|\n",
      "|          5|    2|        1.06865275E8|\n",
      "|          5|    3| 9.263686774193548E7|\n",
      "|          5|    4|1.0065214333333333E8|\n",
      "|          5|    5| 9.880788064516129E7|\n",
      "|          5|    6| 8.147986333333333E7|\n",
      "|          5|    7| 8.913521333333333E7|\n",
      "|          5|    8| 8.461566333333333E7|\n",
      "|          5|    9| 9.657708387096775E7|\n",
      "|          5|   10| 8.505357096774194E7|\n",
      "|          5|   11| 9.265071363636364E7|\n",
      "|          5|   12| 7.037784516129032E7|\n",
      "|          6|    1|         1.1184468E8|\n",
      "|          6|    2| 9.653823571428572E7|\n",
      "|          6|    3|1.0623423571428572E8|\n",
      "|          6|    4|1.0818935384615384E8|\n",
      "|          6|    5|1.0139194193548387E8|\n",
      "|          6|    6|           9.29633E7|\n",
      "|          6|    7|          8.421862E7|\n",
      "|          6|    8| 8.431098709677419E7|\n",
      "|          6|    9| 9.598559666666667E7|\n",
      "|          6|   10| 9.237404516129032E7|\n",
      "|          6|   11|          8.166534E7|\n",
      "|          6|   12| 7.835513793103448E7|\n",
      "+-----------+-----+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df5=df.withColumn('Day-of-Week',dayofweek('Date')).\\\n",
    "withColumn('Month',month('Date')).\\\n",
    "select(['Day-of-Week','Month','Volume']).\\\n",
    "groupby(['Day-of-Week','Month']).\\\n",
    "mean().\\\n",
    "select(['Day-of-Week','Month','avg(Volume)']).\\\n",
    "orderBy(['Day-of-Week','Month'])\n",
    "\n",
    "df5.show(60)"
   ]
  }
 ],
 "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.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}