Notebooks/Spark-Example-12-Taxi Dataset Data Frame Example.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "64k1f686TlpX"
      },
      "source": [
        "# Taxi Dataset Data Frame Example\n",
        "This example is based on the NYC Taxi Dataset and shows how to use the Spark Data Frame API to make basic data analysis."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "24L8SpQe5aEY"
      },
      "outputs": [],
      "source": [
        "import findspark\n",
        "findspark.init()\n",
        "from pyspark import SparkContext\n",
        "from pyspark.sql import SparkSession\n",
        "sc = SparkContext(\"local\")\n",
        "spark = SparkSession.builder.getOrCreate()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {},
      "outputs": [],
      "source": [
        "#spark.stop()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Load the data"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {},
      "outputs": [],
      "source": [
        "def correctRows(p):\n",
        "    if(len(p)==17):\n",
        "        if(isfloat(p[5]) and isfloat(p[11])):\n",
        "            if(float(p[4])> 60 and float(p[5])>0.10 and float(p[11])> 0.10 and float(p[16])> 0.10):\n",
        "                return p\n",
        "#Exception Handling and removing wrong datalines\n",
        "def isfloat(value):\n",
        "    try:\n",
        "        float(value)\n",
        "        return True\n",
        "    except:\n",
        "         return False"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "abZiG_jJ5wHV",
        "outputId": "51e87322-b53d-4c89-9b8b-6ada1b6c899b"
      },
      "outputs": [],
      "source": [
        "path=\"data\\\\\"\n",
        "testFile= path + \"taxi-data-sorted-verysmall.csv\"\n",
        "df = spark.read.format('csv').options(header='false', inferSchema='true',  sep =\",\").load(testFile)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {},
      "outputs": [
        {
          "data": {
            "text/plain": [
              "10000"
            ]
          },
          "execution_count": 4,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "df.count()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+--------------------+--------------------+-------------------+-------------------+---+----+----------+---------+----------+---------+----+----+----+----+----+----+----+\n",
            "|                 _c0|                 _c1|                _c2|                _c3|_c4| _c5|       _c6|      _c7|       _c8|      _c9|_c10|_c11|_c12|_c13|_c14|_c15|_c16|\n",
            "+--------------------+--------------------+-------------------+-------------------+---+----+----------+---------+----------+---------+----+----+----+----+----+----+----+\n",
            "|07290D3599E7A0D62...|E7750A37CAB07D0DF...|2013-01-01 00:00:00|2013-01-01 00:02:00|120|0.44|-73.956528|40.716976| -73.96244|40.715008| CSH| 3.5| 0.5| 0.5| 0.0| 0.0| 4.5|\n",
            "|22D70BF00EEB0ADC8...|3FF2709163DE7036F...|2013-01-01 00:02:00|2013-01-01 00:02:00|  0| 0.0|       0.0|      0.0|       0.0|      0.0| CSH|27.0| 0.0| 0.5| 0.0| 0.0|27.5|\n",
            "|0EC22AAF491A8BD91...|778C92B26AE78A9EB...|2013-01-01 00:01:00|2013-01-01 00:03:00|120|0.71|-73.973145|40.752827|-73.965897|40.760445| CSH| 4.0| 0.5| 0.5| 0.0| 0.0| 5.0|\n",
            "|1390FB380189DF6BB...|BE317B986700F63C4...|2013-01-01 00:01:00|2013-01-01 00:03:00|120|0.48|-74.004173|40.720947|-74.003838|40.726189| CSH| 4.0| 0.5| 0.5| 0.0| 0.0| 5.0|\n",
            "|3B4129883A1D05BE8...|7077F9FD5AD649AEA...|2013-01-01 00:01:00|2013-01-01 00:03:00|120|0.61|-73.987373|40.724861|-73.983772|40.730995| CRD| 4.0| 0.5| 0.5| 0.0| 0.0| 5.0|\n",
            "+--------------------+--------------------+-------------------+-------------------+---+----+----------+---------+----------+---------+----+----+----+----+----+----+----+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "df.show(5, truncate=True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Convert to Pandas DataFrame\n",
        "Converting a Spark DataFrame to a Pandas DataFrame is a common operation that can be particularly useful when you need to perform tasks such as data visualization or analysis that are more convenient in the Pandas ecosystem. One significant advantage of using Pandas DataFrames is their visually appealing and interactive display when used within Jupyter Notebooks."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {},
      "outputs": [
        {
          "data": {
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>_c0</th>\n",
              "      <th>_c1</th>\n",
              "      <th>_c2</th>\n",
              "      <th>_c3</th>\n",
              "      <th>_c4</th>\n",
              "      <th>_c5</th>\n",
              "      <th>_c6</th>\n",
              "      <th>_c7</th>\n",
              "      <th>_c8</th>\n",
              "      <th>_c9</th>\n",
              "      <th>_c10</th>\n",
              "      <th>_c11</th>\n",
              "      <th>_c12</th>\n",
              "      <th>_c13</th>\n",
              "      <th>_c14</th>\n",
              "      <th>_c15</th>\n",
              "      <th>_c16</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>07290D3599E7A0D62097A346EFCC1FB5</td>\n",
              "      <td>E7750A37CAB07D0DFF0AF7E3573AC141</td>\n",
              "      <td>2013-01-01 00:00:00</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.44</td>\n",
              "      <td>-73.956528</td>\n",
              "      <td>40.716976</td>\n",
              "      <td>-73.962440</td>\n",
              "      <td>40.715008</td>\n",
              "      <td>CSH</td>\n",
              "      <td>3.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>4.5</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>22D70BF00EEB0ADC83BA8177BB861991</td>\n",
              "      <td>3FF2709163DE7036FCAA4E5A3324E4BF</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>0</td>\n",
              "      <td>0.00</td>\n",
              "      <td>0.000000</td>\n",
              "      <td>0.000000</td>\n",
              "      <td>0.000000</td>\n",
              "      <td>0.000000</td>\n",
              "      <td>CSH</td>\n",
              "      <td>27.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>27.5</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>0EC22AAF491A8BD91F279350C2B010FD</td>\n",
              "      <td>778C92B26AE78A9EBDF96B49C67E4007</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.71</td>\n",
              "      <td>-73.973145</td>\n",
              "      <td>40.752827</td>\n",
              "      <td>-73.965897</td>\n",
              "      <td>40.760445</td>\n",
              "      <td>CSH</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>1390FB380189DF6BBFDA4DC847CAD14F</td>\n",
              "      <td>BE317B986700F63C43438482792C8654</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.48</td>\n",
              "      <td>-74.004173</td>\n",
              "      <td>40.720947</td>\n",
              "      <td>-74.003838</td>\n",
              "      <td>40.726189</td>\n",
              "      <td>CSH</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>3B4129883A1D05BE89F2C929DE136281</td>\n",
              "      <td>7077F9FD5AD649AEACA4746B2537E3FA</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.61</td>\n",
              "      <td>-73.987373</td>\n",
              "      <td>40.724861</td>\n",
              "      <td>-73.983772</td>\n",
              "      <td>40.730995</td>\n",
              "      <td>CRD</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                _c0                               _c1   \n",
              "0  07290D3599E7A0D62097A346EFCC1FB5  E7750A37CAB07D0DFF0AF7E3573AC141  \\\n",
              "1  22D70BF00EEB0ADC83BA8177BB861991  3FF2709163DE7036FCAA4E5A3324E4BF   \n",
              "2  0EC22AAF491A8BD91F279350C2B010FD  778C92B26AE78A9EBDF96B49C67E4007   \n",
              "3  1390FB380189DF6BBFDA4DC847CAD14F  BE317B986700F63C43438482792C8654   \n",
              "4  3B4129883A1D05BE89F2C929DE136281  7077F9FD5AD649AEACA4746B2537E3FA   \n",
              "\n",
              "                   _c2                  _c3  _c4   _c5        _c6        _c7   \n",
              "0  2013-01-01 00:00:00  2013-01-01 00:02:00  120  0.44 -73.956528  40.716976  \\\n",
              "1  2013-01-01 00:02:00  2013-01-01 00:02:00    0  0.00   0.000000   0.000000   \n",
              "2  2013-01-01 00:01:00  2013-01-01 00:03:00  120  0.71 -73.973145  40.752827   \n",
              "3  2013-01-01 00:01:00  2013-01-01 00:03:00  120  0.48 -74.004173  40.720947   \n",
              "4  2013-01-01 00:01:00  2013-01-01 00:03:00  120  0.61 -73.987373  40.724861   \n",
              "\n",
              "         _c8        _c9 _c10  _c11  _c12  _c13  _c14  _c15  _c16  \n",
              "0 -73.962440  40.715008  CSH   3.5   0.5   0.5   0.0   0.0   4.5  \n",
              "1   0.000000   0.000000  CSH  27.0   0.0   0.5   0.0   0.0  27.5  \n",
              "2 -73.965897  40.760445  CSH   4.0   0.5   0.5   0.0   0.0   5.0  \n",
              "3 -74.003838  40.726189  CSH   4.0   0.5   0.5   0.0   0.0   5.0  \n",
              "4 -73.983772  40.730995  CRD   4.0   0.5   0.5   0.0   0.0   5.0  "
            ]
          },
          "execution_count": 6,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "import pandas as pd\n",
        "# Convert to Pandas Data Frame\n",
        "df.toPandas().head(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Print the schema"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "8MufUYlGN0Sx",
        "outputId": "e0502b69-b417-4eb7-f115-b6efcc965ef9"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- _c0: string (nullable = true)\n",
            " |-- _c1: string (nullable = true)\n",
            " |-- _c2: string (nullable = true)\n",
            " |-- _c3: string (nullable = true)\n",
            " |-- _c4: integer (nullable = true)\n",
            " |-- _c5: double (nullable = true)\n",
            " |-- _c6: double (nullable = true)\n",
            " |-- _c7: double (nullable = true)\n",
            " |-- _c8: double (nullable = true)\n",
            " |-- _c9: double (nullable = true)\n",
            " |-- _c10: string (nullable = true)\n",
            " |-- _c11: double (nullable = true)\n",
            " |-- _c12: double (nullable = true)\n",
            " |-- _c13: double (nullable = true)\n",
            " |-- _c14: double (nullable = true)\n",
            " |-- _c15: double (nullable = true)\n",
            " |-- _c16: double (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Print Schema\n",
        "df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Rename columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- medallion: string (nullable = true)\n",
            " |-- hack_license: string (nullable = true)\n",
            " |-- pickup_datetime: string (nullable = true)\n",
            " |-- dropoff_datetime: string (nullable = true)\n",
            " |-- trip_time: integer (nullable = true)\n",
            " |-- trip_distance: double (nullable = true)\n",
            " |-- pickup_longitude: double (nullable = true)\n",
            " |-- pickup_latitude: double (nullable = true)\n",
            " |-- dropoff_longitude: double (nullable = true)\n",
            " |-- dropoff_latitude: double (nullable = true)\n",
            " |-- payment_type: string (nullable = true)\n",
            " |-- fare_amount: double (nullable = true)\n",
            " |-- surcharge: double (nullable = true)\n",
            " |-- mta_tax: double (nullable = true)\n",
            " |-- tip_amount: double (nullable = true)\n",
            " |-- tolls_amount: double (nullable = true)\n",
            " |-- total_amount: double (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "colum_names = [\"medallion\",\n",
        "    \"hack_license\",\n",
        "    'pickup_datetime',\n",
        "    'dropoff_datetime',\n",
        "    'trip_time',\n",
        "    'trip_distance',\n",
        "    'pickup_longitude',\n",
        "    'pickup_latitude',\n",
        "    'dropoff_longitude',\n",
        "    'dropoff_latitude',\n",
        "    'payment_type',\n",
        "    'fare_amount',\n",
        "    'surcharge',\n",
        "    'mta_tax',\n",
        "    'tip_amount',\n",
        "    'tolls_amount',\n",
        "    'total_amount',\n",
        "] \n",
        "df = df.toDF(*colum_names)\n",
        "df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Drop Columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- medallion: string (nullable = true)\n",
            " |-- hack_license: string (nullable = true)\n",
            " |-- pickup_datetime: string (nullable = true)\n",
            " |-- dropoff_datetime: string (nullable = true)\n",
            " |-- trip_time: integer (nullable = true)\n",
            " |-- trip_distance: double (nullable = true)\n",
            " |-- payment_type: string (nullable = true)\n",
            " |-- fare_amount: double (nullable = true)\n",
            " |-- surcharge: double (nullable = true)\n",
            " |-- mta_tax: double (nullable = true)\n",
            " |-- tip_amount: double (nullable = true)\n",
            " |-- tolls_amount: double (nullable = true)\n",
            " |-- total_amount: double (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "drop_colum_names = [\n",
        "    'pickup_longitude',\n",
        "    'pickup_latitude',\n",
        "    'dropoff_longitude',\n",
        "    'dropoff_latitude',\n",
        "]\n",
        "df = df.drop(*drop_colum_names) \n",
        "df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Add new columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6zevLIm75wKA",
        "outputId": "4342cef2-3d96-4d1b-a0bc-a61b02f7e8a8"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+--------------------------------+--------------------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+\n",
            "|medallion                       |hack_license                    |pickup_datetime    |dropoff_datetime   |trip_time|trip_distance|payment_type|fare_amount|surcharge|mta_tax|tip_amount|tolls_amount|total_amount|year|month|day|hour|minute|\n",
            "+--------------------------------+--------------------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+\n",
            "|07290D3599E7A0D62097A346EFCC1FB5|E7750A37CAB07D0DFF0AF7E3573AC141|2013-01-01 00:00:00|2013-01-01 00:02:00|120      |0.44         |CSH         |3.5        |0.5      |0.5    |0.0       |0.0         |4.5         |2013|01   |01 |00  |00    |\n",
            "|22D70BF00EEB0ADC83BA8177BB861991|3FF2709163DE7036FCAA4E5A3324E4BF|2013-01-01 00:02:00|2013-01-01 00:02:00|0        |0.0          |CSH         |27.0       |0.0      |0.5    |0.0       |0.0         |27.5        |2013|01   |01 |00  |02    |\n",
            "|0EC22AAF491A8BD91F279350C2B010FD|778C92B26AE78A9EBDF96B49C67E4007|2013-01-01 00:01:00|2013-01-01 00:03:00|120      |0.71         |CSH         |4.0        |0.5      |0.5    |0.0       |0.0         |5.0         |2013|01   |01 |00  |01    |\n",
            "|1390FB380189DF6BBFDA4DC847CAD14F|BE317B986700F63C43438482792C8654|2013-01-01 00:01:00|2013-01-01 00:03:00|120      |0.48         |CSH         |4.0        |0.5      |0.5    |0.0       |0.0         |5.0         |2013|01   |01 |00  |01    |\n",
            "|3B4129883A1D05BE89F2C929DE136281|7077F9FD5AD649AEACA4746B2537E3FA|2013-01-01 00:01:00|2013-01-01 00:03:00|120      |0.61         |CRD         |4.0        |0.5      |0.5    |0.0       |0.0         |5.0         |2013|01   |01 |00  |01    |\n",
            "+--------------------------------+--------------------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "from pyspark.sql.functions import col, substring\n",
        "# Note: Please note that the position is not zero based, but 1 based index.\n",
        "# Create a new columns wiht the year, month, day, hour and minute of the taxi trip\n",
        "df = df.withColumn('year', substring('pickup_datetime', 1,4))\\\n",
        "    .withColumn('month', substring('pickup_datetime', 6,2))\\\n",
        "    .withColumn('day', substring('pickup_datetime', 9,2))\\\n",
        "    .withColumn('hour', substring('pickup_datetime', 12,2))\\\n",
        "    .withColumn('minute', substring('pickup_datetime', 15,2))\n",
        "df.show(5,truncate=False, vertical=False)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {},
      "outputs": [
        {
          "data": {
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>medallion</th>\n",
              "      <th>hack_license</th>\n",
              "      <th>pickup_datetime</th>\n",
              "      <th>dropoff_datetime</th>\n",
              "      <th>trip_time</th>\n",
              "      <th>trip_distance</th>\n",
              "      <th>payment_type</th>\n",
              "      <th>fare_amount</th>\n",
              "      <th>surcharge</th>\n",
              "      <th>mta_tax</th>\n",
              "      <th>tip_amount</th>\n",
              "      <th>tolls_amount</th>\n",
              "      <th>total_amount</th>\n",
              "      <th>year</th>\n",
              "      <th>month</th>\n",
              "      <th>day</th>\n",
              "      <th>hour</th>\n",
              "      <th>minute</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>07290D3599E7A0D62097A346EFCC1FB5</td>\n",
              "      <td>E7750A37CAB07D0DFF0AF7E3573AC141</td>\n",
              "      <td>2013-01-01 00:00:00</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.44</td>\n",
              "      <td>CSH</td>\n",
              "      <td>3.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>4.5</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>00</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>22D70BF00EEB0ADC83BA8177BB861991</td>\n",
              "      <td>3FF2709163DE7036FCAA4E5A3324E4BF</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>2013-01-01 00:02:00</td>\n",
              "      <td>0</td>\n",
              "      <td>0.00</td>\n",
              "      <td>CSH</td>\n",
              "      <td>27.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>27.5</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>02</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>0EC22AAF491A8BD91F279350C2B010FD</td>\n",
              "      <td>778C92B26AE78A9EBDF96B49C67E4007</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.71</td>\n",
              "      <td>CSH</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>01</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>1390FB380189DF6BBFDA4DC847CAD14F</td>\n",
              "      <td>BE317B986700F63C43438482792C8654</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.48</td>\n",
              "      <td>CSH</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>01</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>3B4129883A1D05BE89F2C929DE136281</td>\n",
              "      <td>7077F9FD5AD649AEACA4746B2537E3FA</td>\n",
              "      <td>2013-01-01 00:01:00</td>\n",
              "      <td>2013-01-01 00:03:00</td>\n",
              "      <td>120</td>\n",
              "      <td>0.61</td>\n",
              "      <td>CRD</td>\n",
              "      <td>4.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>5.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>01</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>...</th>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "      <td>...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>9995</th>\n",
              "      <td>F476E33AE955F276113772AB44540488</td>\n",
              "      <td>B7C3EDB805D5074CFFBA649A98CBCC3D</td>\n",
              "      <td>2013-01-01 00:44:00</td>\n",
              "      <td>2013-01-01 00:56:00</td>\n",
              "      <td>720</td>\n",
              "      <td>2.22</td>\n",
              "      <td>CSH</td>\n",
              "      <td>10.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>11.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>44</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>9996</th>\n",
              "      <td>F6AC5ED8F624EC43788EA84D9979379C</td>\n",
              "      <td>241AB3536C3494045D967E758844F5B6</td>\n",
              "      <td>2013-01-01 00:49:00</td>\n",
              "      <td>2013-01-01 00:56:00</td>\n",
              "      <td>420</td>\n",
              "      <td>0.75</td>\n",
              "      <td>CRD</td>\n",
              "      <td>6.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.0</td>\n",
              "      <td>8.3</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>49</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>9997</th>\n",
              "      <td>F865789AEA73DF9A5CDECE38BC5BB3B5</td>\n",
              "      <td>2223D379E6348801232994F06477548E</td>\n",
              "      <td>2013-01-01 00:41:00</td>\n",
              "      <td>2013-01-01 00:56:00</td>\n",
              "      <td>900</td>\n",
              "      <td>3.64</td>\n",
              "      <td>CRD</td>\n",
              "      <td>14.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>1.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>16.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>41</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>9998</th>\n",
              "      <td>F920D6FCA90502CC780348B1AC364596</td>\n",
              "      <td>DB93A24BCE932FD908322C103C821A39</td>\n",
              "      <td>2013-01-01 00:42:00</td>\n",
              "      <td>2013-01-01 00:56:00</td>\n",
              "      <td>840</td>\n",
              "      <td>1.43</td>\n",
              "      <td>CSH</td>\n",
              "      <td>10.0</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>11.0</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>42</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>9999</th>\n",
              "      <td>FAC826F3A42DAC858AAB54B7509D7AC7</td>\n",
              "      <td>5DDBB30521FB88F3E242DEEEAFB47F4B</td>\n",
              "      <td>2013-01-01 00:43:00</td>\n",
              "      <td>2013-01-01 00:56:00</td>\n",
              "      <td>780</td>\n",
              "      <td>2.28</td>\n",
              "      <td>CSH</td>\n",
              "      <td>11.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.5</td>\n",
              "      <td>0.0</td>\n",
              "      <td>0.0</td>\n",
              "      <td>12.5</td>\n",
              "      <td>2013</td>\n",
              "      <td>01</td>\n",
              "      <td>01</td>\n",
              "      <td>00</td>\n",
              "      <td>43</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "<p>10000 rows × 18 columns</p>\n",
              "</div>"
            ],
            "text/plain": [
              "                             medallion                      hack_license   \n",
              "0     07290D3599E7A0D62097A346EFCC1FB5  E7750A37CAB07D0DFF0AF7E3573AC141  \\\n",
              "1     22D70BF00EEB0ADC83BA8177BB861991  3FF2709163DE7036FCAA4E5A3324E4BF   \n",
              "2     0EC22AAF491A8BD91F279350C2B010FD  778C92B26AE78A9EBDF96B49C67E4007   \n",
              "3     1390FB380189DF6BBFDA4DC847CAD14F  BE317B986700F63C43438482792C8654   \n",
              "4     3B4129883A1D05BE89F2C929DE136281  7077F9FD5AD649AEACA4746B2537E3FA   \n",
              "...                                ...                               ...   \n",
              "9995  F476E33AE955F276113772AB44540488  B7C3EDB805D5074CFFBA649A98CBCC3D   \n",
              "9996  F6AC5ED8F624EC43788EA84D9979379C  241AB3536C3494045D967E758844F5B6   \n",
              "9997  F865789AEA73DF9A5CDECE38BC5BB3B5  2223D379E6348801232994F06477548E   \n",
              "9998  F920D6FCA90502CC780348B1AC364596  DB93A24BCE932FD908322C103C821A39   \n",
              "9999  FAC826F3A42DAC858AAB54B7509D7AC7  5DDBB30521FB88F3E242DEEEAFB47F4B   \n",
              "\n",
              "          pickup_datetime     dropoff_datetime  trip_time  trip_distance   \n",
              "0     2013-01-01 00:00:00  2013-01-01 00:02:00        120           0.44  \\\n",
              "1     2013-01-01 00:02:00  2013-01-01 00:02:00          0           0.00   \n",
              "2     2013-01-01 00:01:00  2013-01-01 00:03:00        120           0.71   \n",
              "3     2013-01-01 00:01:00  2013-01-01 00:03:00        120           0.48   \n",
              "4     2013-01-01 00:01:00  2013-01-01 00:03:00        120           0.61   \n",
              "...                   ...                  ...        ...            ...   \n",
              "9995  2013-01-01 00:44:00  2013-01-01 00:56:00        720           2.22   \n",
              "9996  2013-01-01 00:49:00  2013-01-01 00:56:00        420           0.75   \n",
              "9997  2013-01-01 00:41:00  2013-01-01 00:56:00        900           3.64   \n",
              "9998  2013-01-01 00:42:00  2013-01-01 00:56:00        840           1.43   \n",
              "9999  2013-01-01 00:43:00  2013-01-01 00:56:00        780           2.28   \n",
              "\n",
              "     payment_type  fare_amount  surcharge  mta_tax  tip_amount  tolls_amount   \n",
              "0             CSH          3.5        0.5      0.5         0.0           0.0  \\\n",
              "1             CSH         27.0        0.0      0.5         0.0           0.0   \n",
              "2             CSH          4.0        0.5      0.5         0.0           0.0   \n",
              "3             CSH          4.0        0.5      0.5         0.0           0.0   \n",
              "4             CRD          4.0        0.5      0.5         0.0           0.0   \n",
              "...           ...          ...        ...      ...         ...           ...   \n",
              "9995          CSH         10.0        0.5      0.5         0.0           0.0   \n",
              "9996          CRD          6.0        0.5      0.5         1.3           0.0   \n",
              "9997          CRD         14.0        0.5      0.5         1.0           0.0   \n",
              "9998          CSH         10.0        0.5      0.5         0.0           0.0   \n",
              "9999          CSH         11.5        0.5      0.5         0.0           0.0   \n",
              "\n",
              "      total_amount  year month day hour minute  \n",
              "0              4.5  2013    01  01   00     00  \n",
              "1             27.5  2013    01  01   00     02  \n",
              "2              5.0  2013    01  01   00     01  \n",
              "3              5.0  2013    01  01   00     01  \n",
              "4              5.0  2013    01  01   00     01  \n",
              "...            ...   ...   ...  ..  ...    ...  \n",
              "9995          11.0  2013    01  01   00     44  \n",
              "9996           8.3  2013    01  01   00     49  \n",
              "9997          16.0  2013    01  01   00     41  \n",
              "9998          11.0  2013    01  01   00     42  \n",
              "9999          12.5  2013    01  01   00     43  \n",
              "\n",
              "[10000 rows x 18 columns]"
            ]
          },
          "execution_count": 11,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "# More beautiful way to show the data frame\n",
        "# Note: Use this only for small data frames. \n",
        "# This operation will collect all data to the driver and can lead to out of memory errors \n",
        "# or very slow performance.\n",
        "df.toPandas()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+--------------------+--------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+-------------+\n",
            "|           medallion|        hack_license|    pickup_datetime|   dropoff_datetime|trip_time|trip_distance|payment_type|fare_amount|surcharge|mta_tax|tip_amount|tolls_amount|total_amount|year|month|day|hour|minute|fare_category|\n",
            "+--------------------+--------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+-------------+\n",
            "|07290D3599E7A0D62...|E7750A37CAB07D0DF...|2013-01-01 00:00:00|2013-01-01 00:02:00|      120|         0.44|         CSH|        3.5|      0.5|    0.5|       0.0|         0.0|         4.5|2013|   01| 01|  00|    00|            3|\n",
            "|22D70BF00EEB0ADC8...|3FF2709163DE7036F...|2013-01-01 00:02:00|2013-01-01 00:02:00|        0|          0.0|         CSH|       27.0|      0.0|    0.5|       0.0|         0.0|        27.5|2013|   01| 01|  00|    02|           27|\n",
            "|0EC22AAF491A8BD91...|778C92B26AE78A9EB...|2013-01-01 00:01:00|2013-01-01 00:03:00|      120|         0.71|         CSH|        4.0|      0.5|    0.5|       0.0|         0.0|         5.0|2013|   01| 01|  00|    01|            4|\n",
            "|1390FB380189DF6BB...|BE317B986700F63C4...|2013-01-01 00:01:00|2013-01-01 00:03:00|      120|         0.48|         CSH|        4.0|      0.5|    0.5|       0.0|         0.0|         5.0|2013|   01| 01|  00|    01|            4|\n",
            "|3B4129883A1D05BE8...|7077F9FD5AD649AEA...|2013-01-01 00:01:00|2013-01-01 00:03:00|      120|         0.61|         CRD|        4.0|      0.5|    0.5|       0.0|         0.0|         5.0|2013|   01| 01|  00|    01|            4|\n",
            "+--------------------+--------------------+-------------------+-------------------+---------+-------------+------------+-----------+---------+-------+----------+------------+------------+----+-----+---+----+------+-------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Create fare_category column as a integer of the fare_amount\n",
        "df = df.withColumn('fare_category', df['fare_amount'].cast(\"int\"))\n",
        "df.show(5)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-------------+-----+\n",
            "|fare_category|count|\n",
            "+-------------+-----+\n",
            "|           31|   13|\n",
            "|           53|    2|\n",
            "|           34|   12|\n",
            "|           28|   38|\n",
            "|           27|   46|\n",
            "|           26|   51|\n",
            "|           44|    2|\n",
            "|           12|  416|\n",
            "|           22|  118|\n",
            "|          259|    1|\n",
            "+-------------+-----+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# find the number of trips for each fare category\n",
        "df.groupBy('fare_category').count().show(10)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "-jzBJTZGOUXf",
        "outputId": "eeb7b046-092b-48bc-df40-6d97656f0682"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+------+-----+\n",
            "|minute|count|\n",
            "+------+-----+\n",
            "|    07|  165|\n",
            "|    51|   44|\n",
            "|    15|  231|\n",
            "|    54|   10|\n",
            "|    11|  221|\n",
            "+------+-----+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Find the number of trips in given minute of the hour\n",
        "df.groupBy('minute').count().show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Sort and limit"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-------------+-----+\n",
            "|fare_category|count|\n",
            "+-------------+-----+\n",
            "|            6| 1066|\n",
            "|            7|  993|\n",
            "|            5|  910|\n",
            "|            8|  847|\n",
            "|            9|  772|\n",
            "+-------------+-----+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Find the top 5 fare categories with the highest number of trips\n",
        "df.groupBy('fare_category').count().orderBy(['count'],ascending=False).limit(5).show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Convert to RDD \n",
        "The data frame can be converted to an RDD using the rdd method. The resulting RDD will be an RDD of Row objects. Row objects are similar to dictionaries in that they have keys and values. In this case, the keys are the column names, and the values are the entries in the corresponding columns."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 17,
      "metadata": {},
      "outputs": [
        {
          "data": {
            "text/plain": [
              "[Row(minute='07', count=165),\n",
              " Row(minute='51', count=44),\n",
              " Row(minute='15', count=231),\n",
              " Row(minute='54', count=10),\n",
              " Row(minute='11', count=221)]"
            ]
          },
          "execution_count": 17,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "r = df.groupBy('minute').count().rdd\n",
        "r.take(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "The more convenient way is to convert rows to tuples and then do the data analysis."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Dwn6mfE-5wMu",
        "outputId": "e695036f-3528-4c3f-a8ef-a6bc7dc07155"
      },
      "outputs": [
        {
          "data": {
            "text/plain": [
              "[('07', 165), ('51', 44), ('15', 231), ('54', 10), ('11', 221)]"
            ]
          },
          "execution_count": 18,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "# convert to RDD\n",
        "r = df.groupBy('minute').count().rdd.map(tuple)\n",
        "r.take(5)"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "name": "Spark-Example-09a-treeAggregate.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3.10.4 ('spark')",
      "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.10.14"
    },
    "vscode": {
      "interpreter": {
        "hash": "9db6cbf0fd79f8e79653fe7b0c50b956ca6e525ee712295da3c66f75e4fe96ce"
      }
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}