GroupBy_and_Aggregate_Functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# GroupBy and Aggregate Functions\n",
    "\n",
    "GroupBy allows you to group rows together based off some column value, for example, you could group together sales data by the day the sale occured, or group repeast customer data based off the name of the customer. Once you've performed the GroupBy operation you can use an aggregate function off that data. An aggregate function aggregates multiple rows of data into a single output, such as taking the sum of inputs, or counting the number of inputs.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# May take a little while on a local computer\n",
    "spark = SparkSession.builder.appName(\"groupbyagg\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read in the customer sales data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv('sales_info.csv',inferSchema=True,header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "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": 8,
   "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": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's group together by company!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.group.GroupedData at 0x109915f28>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(\"Company\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This returns a GroupedData object, off of which you can all various methods"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "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": [
    "# Mean\n",
    "df.groupBy(\"Company\").mean().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "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": [
    "# Count\n",
    "df.groupBy(\"Company\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "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": [
    "# Max\n",
    "df.groupBy(\"Company\").max().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "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": [
    "# Min\n",
    "df.groupBy(\"Company\").min().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+\n",
      "|Company|sum(Sales)|\n",
      "+-------+----------+\n",
      "|   APPL|    1480.0|\n",
      "|   GOOG|     660.0|\n",
      "|     FB|    1220.0|\n",
      "|   MSFT|     967.0|\n",
      "+-------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Sum\n",
    "df.groupBy(\"Company\").sum().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check out this link for more info on other methods:\n",
    "http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark-sql-module\n",
    "\n",
    "Not all methods need a groupby call, instead one can just call the generalized **.agg()** method, that will call the aggregate across all rows in the dataframe column specified. It can take in arguments as a single column, or create multiple aggregate calls all at once using dictionary notation.\n",
    "\n",
    "For example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|max(Sales)|\n",
      "+----------+\n",
      "|     870.0|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Max sales across everything\n",
    "df.agg({'Sales':'max'}).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Could have done this on the group by object as well:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "grouped = df.groupBy(\"Company\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "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": [
    "grouped.agg({\"Sales\":'max'}).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Functions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import countDistinct, avg,stddev"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|count(DISTINCT Sales)|\n",
      "+---------------------+\n",
      "|                   11|\n",
      "+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(countDistinct(\"Sales\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Often you will want to change the name, use the .alias() method for this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Distinct Sales|\n",
      "+--------------+\n",
      "|            11|\n",
      "+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(countDistinct(\"Sales\").alias(\"Distinct Sales\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "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": 38,
   "metadata": {},
   "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": 39,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import format_number"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sales_std = df.select(stddev(\"Sales\").alias('std'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "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": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|format_number(std, 2)|\n",
      "+---------------------+\n",
      "|               250.09|\n",
      "+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# format_number(\"col_name\",decimal places)\n",
    "sales_std.select(format_number('std',2)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Order By\n",
    "\n",
    "You can easily sort with the orderBy method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "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": [
    "# OrderBy\n",
    "# Ascending\n",
    "df.orderBy(\"Sales\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "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": [
    "# Descending call off the column itself.\n",
    "df.orderBy(df[\"Sales\"].desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Most basic functions you would expect to be available are, so make sure to check out the documentation!"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}