Home_sales.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "sIT9amuYt-5m",
        "outputId": "0130058c-6384-4612-fe0c-5eda8b692f92"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rHit:1 https://cloud.r-project.org/bin/linux/ubuntu jammy-cran40/ InRelease\n",
            "\r0% [Connecting to archive.ubuntu.com (185.125.190.82)] [Connecting to security.ubuntu.com (185.125.1\r                                                                                                    \rHit:2 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2204/x86_64  InRelease\n",
            "\r0% [Connecting to archive.ubuntu.com (185.125.190.82)] [Connecting to security.ubuntu.com (185.125.1\r0% [Waiting for headers] [Waiting for headers] [Waiting for headers] [Connecting to ppa.launchpadcon\r                                                                                                    \rIgn:3 https://r2u.stat.illinois.edu/ubuntu jammy InRelease\n",
            "\r0% [Waiting for headers] [Waiting for headers] [Connected to ppa.launchpadcontent.net (185.125.190.8\r0% [Waiting for headers] [Waiting for headers] [Waiting for headers] [Connected to ppa.launchpadcont\r                                                                                                    \rHit:4 http://archive.ubuntu.com/ubuntu jammy InRelease\n",
            "Hit:5 http://security.ubuntu.com/ubuntu jammy-security InRelease\n",
            "Hit:6 https://r2u.stat.illinois.edu/ubuntu jammy Release\n",
            "Hit:7 http://archive.ubuntu.com/ubuntu jammy-updates InRelease\n",
            "Hit:8 http://archive.ubuntu.com/ubuntu jammy-backports InRelease\n",
            "Hit:10 https://ppa.launchpadcontent.net/deadsnakes/ppa/ubuntu jammy InRelease\n",
            "Hit:11 https://ppa.launchpadcontent.net/graphics-drivers/ppa/ubuntu jammy InRelease\n",
            "Hit:12 https://ppa.launchpadcontent.net/ubuntugis/ppa/ubuntu jammy InRelease\n",
            "Reading package lists... Done\n",
            "W: Skipping acquire of configured file 'main/source/Sources' as repository 'https://r2u.stat.illinois.edu/ubuntu jammy InRelease' does not seem to provide it (sources.list entry misspelt?)\n"
          ]
        }
      ],
      "source": [
        "import os\n",
        "# Define Spark version\n",
        "spark_version = 'spark-3.4.0'\n",
        "os.environ['SPARK_VERSION'] = spark_version\n",
        "\n",
        "# Install Spark and Java\n",
        "!apt-get update\n",
        "!apt-get install openjdk-11-jdk-headless -qq > /dev/null\n",
        "!wget -q https://archive.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop3.tgz\n",
        "!tar xf $SPARK_VERSION-bin-hadoop3.tgz\n",
        "!pip install -q findspark\n",
        "\n",
        "# Set Environment Variables\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = f\"/content/{spark_version}-bin-hadoop3\"\n",
        "\n",
        "# Start a SparkSession\n",
        "import findspark\n",
        "findspark.init()\n"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Import packages\n",
        "from pyspark.sql import SparkSession\n",
        "import time\n",
        "\n",
        "# Create a SparkSession\n",
        "spark = SparkSession.builder.appName(\"SparkSQL\").getOrCreate()"
      ],
      "metadata": {
        "id": "iC-81q9RvYs3"
      },
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 1. Read in the AWS S3 bucket into a DataFrame.\n",
        "from pyspark import SparkFiles\n",
        "url = \"https://2u-data-curriculum-team.s3.amazonaws.com/dataviz-classroom/v1.2/22-big-data/home_sales_revised.csv\"\n",
        "spark.sparkContext.addFile(url)\n",
        "df = spark.read.csv(SparkFiles.get(\"home_sales_revised.csv\"), sep=\",\", header=True, inferSchema=True)"
      ],
      "metadata": {
        "id": "3-L91ATevY6d"
      },
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 2. Create a temporary view of the DataFrame.\n",
        "df.createOrReplaceTempView(\"home_sales\")\n"
      ],
      "metadata": {
        "id": "ilekymyyvZBc"
      },
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 3. What is the average price for a four bedroom house sold per year, rounded to two decimal places?\n",
        "\n",
        "average_price_4_bedroom = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        year(date) as Year,\n",
        "        ROUND(AVG(price), 2) as Average_Price\n",
        "    FROM\n",
        "        home_sales\n",
        "    WHERE\n",
        "        bedrooms == 4\n",
        "    GROUP BY\n",
        "        year(date)\n",
        "    ORDER BY\n",
        "        Year\n",
        "\"\"\")\n",
        "average_price_4_bedroom.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6u0QJMYivZGp",
        "outputId": "0ab22bc8-86d8-465c-e80d-cdbd65c52e81"
      },
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----+-------------+\n",
            "|Year|Average_Price|\n",
            "+----+-------------+\n",
            "|2019|     300263.7|\n",
            "|2020|    298353.78|\n",
            "|2021|    301819.44|\n",
            "|2022|    296363.88|\n",
            "+----+-------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 4. What is the average price of a home for each year the home was built,\n",
        "# that have 3 bedrooms and 3 bathrooms, rounded to two decimal places?\n",
        "\n",
        "average_price_3_bed_3_bath = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        CAST(date_built AS STRING) as Year_Built,\n",
        "        ROUND(AVG(price), 2) as Average_Price\n",
        "    FROM\n",
        "        home_sales\n",
        "    WHERE\n",
        "        bedrooms == 3 AND bathrooms == 3\n",
        "    GROUP BY\n",
        "        Year_Built\n",
        "    ORDER BY\n",
        "        Year_Built\n",
        "\"\"\")\n",
        "average_price_3_bed_3_bath.show()\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "FYol7LVXvZPP",
        "outputId": "a05dd938-64d5-4491-e277-54fba7d71482"
      },
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+-------------+\n",
            "|Year_Built|Average_Price|\n",
            "+----------+-------------+\n",
            "|      2010|    292859.62|\n",
            "|      2011|    291117.47|\n",
            "|      2012|    293683.19|\n",
            "|      2013|    295962.27|\n",
            "|      2014|    290852.27|\n",
            "|      2015|     288770.3|\n",
            "|      2016|    290555.07|\n",
            "|      2017|    292676.79|\n",
            "+----------+-------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 5. What is the average price of a home for each year the home was built,\n",
        "# that have 3 bedrooms, 3 bathrooms, with two floors,\n",
        "# and are greater than or equal to 2,000 square feet, rounded to two decimal places?\n",
        "\n",
        "average_price_filtered = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        CAST(date_built AS STRING) as Year_Built,\n",
        "        ROUND(AVG(price), 2) as Average_Price\n",
        "    FROM\n",
        "        home_sales\n",
        "    WHERE\n",
        "        bedrooms == 3 AND\n",
        "        bathrooms == 3 AND\n",
        "        floors == 2 AND\n",
        "        sqft_living >= 2000\n",
        "    GROUP BY\n",
        "        Year_Built\n",
        "    ORDER BY\n",
        "        Year_Built\n",
        "\"\"\")\n",
        "average_price_filtered.show()\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "idZKJHY9vg8z",
        "outputId": "d70f6cac-2bdf-4380-b73c-b50f761e6327"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+-------------+\n",
            "|Year_Built|Average_Price|\n",
            "+----------+-------------+\n",
            "|      2010|    285010.22|\n",
            "|      2011|    276553.81|\n",
            "|      2012|    307539.97|\n",
            "|      2013|    303676.79|\n",
            "|      2014|    298264.72|\n",
            "|      2015|    297609.97|\n",
            "|      2016|     293965.1|\n",
            "|      2017|    280317.58|\n",
            "+----------+-------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 6. What is the average price of a home per \"view\" rating, rounded to two decimal places,\n",
        "# having an average home price greater than or equal to $350,000? Order by descending view rating.\n",
        "# Although this is a small dataset, determine the run time for this query.\n",
        "\n",
        "start_time = time.time()\n",
        "\n",
        "# Query to calculate the average price per view rating\n",
        "average_price_per_view = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        view AS View_Rating,\n",
        "        ROUND(AVG(price), 2) AS Average_Price\n",
        "    FROM\n",
        "        home_sales\n",
        "    GROUP BY\n",
        "        view\n",
        "    HAVING\n",
        "        AVG(price) >= 350000\n",
        "    ORDER BY\n",
        "        View_Rating DESC\n",
        "\"\"\")\n",
        "\n",
        "\n",
        "print(\"--- %s seconds ---\" % (time.time() - start_time))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "CTIUx9zcvhF6",
        "outputId": "10662a80-e9f4-4ec3-a0d1-4c3c9e55b2aa"
      },
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--- 0.12192535400390625 seconds ---\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 7. Cache the the temporary table home_sales.\n",
        "spark.sql(\"CACHE TABLE home_sales\")"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "-SCFyxyLxpxv",
        "outputId": "a412cef2-349b-41ef-c9e8-dcebcaf814e1"
      },
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "DataFrame[]"
            ]
          },
          "metadata": {},
          "execution_count": 14
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 8. Check if the table is cached.\n",
        "spark.catalog.isCached('home_sales')\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZdyVOK6Exp5n",
        "outputId": "c05f891f-484b-42a4-9e9b-a55641e7f62a"
      },
      "execution_count": 17,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "True"
            ]
          },
          "metadata": {},
          "execution_count": 17
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 9. Using the cached data, run the last query above, that calculates\n",
        "# the average price of a home per \"view\" rating, rounded to two decimal places,\n",
        "# having an average home price greater than or equal to $350,000.\n",
        "# Determine the runtime and compare it to the uncached runtime.\n",
        "\n",
        "start_time = time.time()\n",
        "\n",
        "# Query to calculate the average price per view rating using cached data\n",
        "average_price_per_view_cached = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        view AS View_Rating,\n",
        "        ROUND(AVG(price), 2) AS Average_Price\n",
        "    FROM\n",
        "        home_sales\n",
        "    GROUP BY\n",
        "        view\n",
        "    HAVING\n",
        "        AVG(price) >= 350000\n",
        "    ORDER BY\n",
        "        View_Rating DESC\n",
        "\"\"\")\n",
        "\n",
        "print(\"--- %s seconds ---\" % (time.time() - start_time))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "hbgTQ8suxqBY",
        "outputId": "d111c36f-d914-46dc-8070-5a90458c3104"
      },
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--- 0.05103421211242676 seconds ---\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 10. Partition the data by \"date_built\" and write to a Parquet file\n",
        "df.write.partitionBy(\"date_built\").parquet(\"/content/home_sales_parquet\")\n"
      ],
      "metadata": {
        "id": "1f3OkmkmvhNI"
      },
      "execution_count": 19,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 11. Read the Parquet formatted data\n",
        "parquet_df = spark.read.parquet(\"/content/home_sales_parquet\")\n"
      ],
      "metadata": {
        "id": "K0W_XF7U3RYw"
      },
      "execution_count": 20,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 12. Create a temporary table from the Parquet DataFrame\n",
        "parquet_df.createOrReplaceTempView(\"parquet_home_sales\")\n"
      ],
      "metadata": {
        "id": "H71M2Pz83RdT"
      },
      "execution_count": 21,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 13. Using the parquet DataFrame, run the last query above, that calculates\n",
        "# the average price of a home per \"view\" rating, rounded to two decimal places,\n",
        "# having an average home price greater than or equal to $350,000.\n",
        "# Determine the runtime and compare it to the cached runtime.\n",
        "\n",
        "start_time = time.time()\n",
        "\n",
        "# Query to calculate the average price per view rating using the Parquet data\n",
        "average_price_per_view_parquet = spark.sql(\"\"\"\n",
        "    SELECT\n",
        "        view AS View_Rating,\n",
        "        ROUND(AVG(price), 2) AS Average_Price\n",
        "    FROM\n",
        "        parquet_home_sales\n",
        "    GROUP BY\n",
        "        view\n",
        "    HAVING\n",
        "        AVG(price) >= 350000\n",
        "    ORDER BY\n",
        "        View_Rating DESC\n",
        "\"\"\")\n",
        "\n",
        "\n",
        "print(\"--- %s seconds ---\" % (time.time() - start_time))"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "bWIU8eHY3vvD",
        "outputId": "1a7244f9-4f03-4c4f-836e-87633340c450"
      },
      "execution_count": 22,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--- 0.0701601505279541 seconds ---\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Uncache the home_sales temporary table\n",
        "spark.sql(\"UNCACHE TABLE home_sales\")"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "7ChBxPGB4FHW",
        "outputId": "da29af79-2fce-4034-b8c9-61fbcaaf1614"
      },
      "execution_count": 23,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "DataFrame[]"
            ]
          },
          "metadata": {},
          "execution_count": 23
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "m-1EozUV4FSb"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}