08-Spark-SQL.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "7f41cf64-565f-48b4-a5e5-4d48fc404270",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set the PySpark environment variables\n",
    "import os\n",
    "os.environ['SPARK_HOME'] = \"/Users/coder2j/Apps/Spark\"\n",
    "os.environ['PYSPARK_DRIVER_PYTHON'] = 'jupyter'\n",
    "os.environ['PYSPARK_DRIVER_PYTHON_OPTS'] = 'lab'\n",
    "os.environ['PYSPARK_PYTHON'] = 'python'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "958a52c6-da3d-45b5-9c2e-27e579c6066d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Create a SparkSession\n",
    "spark = SparkSession.builder.appName(\"DataFrameSQL\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "id": "d3a86cbb-583f-427d-970b-de58f9e7bcf4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "name,age,gender,salary\n",
      "John Doe,30,Male,50000\n",
      "Jane Smith,25,Female,45000\n",
      "David Johnson,35,Male,60000\n",
      "Emily Davis,28,Female,52000\n",
      "Michael Wilson,40,Male,75000\n",
      "Sarah Brown,32,Female,58000\n",
      "Robert Lee,29,Male,51000\n",
      "Lisa Garcia,27,Female,49000\n",
      "James Martinez,38,Male,70000\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "head -10 ./data/persons.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24c56d31-71d2-48c0-8d22-b7eb9c4c347f",
   "metadata": {},
   "source": [
    "### Load Data into a DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "dcd9c1d7-ccc2-4cee-8547-c23f3bee42bb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the synthetic data into a DataFrame\n",
    "data_file_path = \"./data/persons.csv\"\n",
    "df = spark.read.csv(data_file_path, header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "id": "99b0d2db-5d0e-4e82-9d45-95029f614319",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- salary: integer (nullable = true)\n",
      "\n",
      "Initial DataFrame:\n",
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|        Jane Smith| 25|Female| 45000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "+------------------+---+------+------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Display schema of DataFrame\n",
    "df.printSchema()\n",
    "\n",
    "# Show the initial DataFrame\n",
    "print(\"Initial DataFrame:\")\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1293db0e-3af1-40d3-9fed-15cb5c6d54f1",
   "metadata": {},
   "source": [
    "### Register the DataFrame as a Temporary Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "id": "439040dd-8c13-48c9-9a72-acd313da407a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register the DataFrame as a Temporary Table\n",
    "df.createOrReplaceTempView(\"my_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3c0ed5db-3985-4251-935a-3c01edd47005",
   "metadata": {},
   "source": [
    "### Perform SQL-like Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "id": "ccf8aa7d-e77f-4fca-93d9-fd4f2581803d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "|  William Anderson| 33|  Male| 62000|\n",
      "|   Karen Hernandez| 31|Female| 55000|\n",
      "|Christopher Taylor| 37|  Male| 69000|\n",
      "|     Matthew Davis| 36|  Male| 67000|\n",
      "|    Patricia White| 29|Female| 50000|\n",
      "|     Daniel Miller| 34|  Male| 64000|\n",
      "| Elizabeth Jackson| 30|Female| 52000|\n",
      "|     Joseph Harris| 28|  Male| 53000|\n",
      "|      Linda Martin| 39|Female| 71000|\n",
      "+------------------+---+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Select all rows where age is greater than 25\n",
    "result = spark.sql(\"SELECT * FROM my_table WHERE age > 25\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "8b88f371-fd06-4aed-9fca-bdb4ce1362ff",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------+\n",
      "|gender|avg_salary|\n",
      "+------+----------+\n",
      "|Female|   52300.0|\n",
      "|  Male|   62100.0|\n",
      "+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Compute the average salary by gender\n",
    "avg_salary_by_gender = spark.sql(\"SELECT gender, AVG(salary) as avg_salary FROM my_table GROUP BY gender\")\n",
    "avg_salary_by_gender.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62b21e82-65e2-4b09-b344-a5cc72d9d430",
   "metadata": {},
   "source": [
    "### Creating and managing temporary views."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "6469c424-87cd-4dfa-b015-1c3f8e591de0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a temporary view\n",
    "df.createOrReplaceTempView(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "id": "9198c861-e79a-4fac-80cc-747c94f062ce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "|  William Anderson| 33|  Male| 62000|\n",
      "|   Karen Hernandez| 31|Female| 55000|\n",
      "|Christopher Taylor| 37|  Male| 69000|\n",
      "|     Matthew Davis| 36|  Male| 67000|\n",
      "|    Patricia White| 29|Female| 50000|\n",
      "|     Daniel Miller| 34|  Male| 64000|\n",
      "| Elizabeth Jackson| 30|Female| 52000|\n",
      "|     Joseph Harris| 28|  Male| 53000|\n",
      "|      Linda Martin| 39|Female| 71000|\n",
      "+------------------+---+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Query the temporary view\n",
    "result = spark.sql(\"SELECT * FROM people WHERE age > 25\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "id": "b66e2493-82c9-4a46-a2c5-049fe9572e7b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check if a temporary view exists\n",
    "view_exists = spark.catalog.tableExists(\"people\")\n",
    "view_exists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "id": "6d363610-7dc6-4a80-bc2d-c1269691e437",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Drop a temporary view\n",
    "spark.catalog.dropTempView(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "id": "38f7e2b9-8d1a-4060-8791-7cbfed9d8b22",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check if a temporary view exists\n",
    "view_exists = spark.catalog.tableExists(\"people\")\n",
    "view_exists"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8bcdf631-ee1d-496f-858b-a13d0b43cb18",
   "metadata": {},
   "source": [
    "### Subquries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "id": "5cc8db71-4862-4169-b2d0-7e38ccdf8331",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "| id|   name|\n",
      "+---+-------+\n",
      "|  1|   John|\n",
      "|  2|  Alice|\n",
      "|  3|    Bob|\n",
      "|  4|  Emily|\n",
      "|  5|  David|\n",
      "|  6|  Sarah|\n",
      "|  7|Michael|\n",
      "|  8|   Lisa|\n",
      "|  9|William|\n",
      "+---+-------+\n",
      "\n",
      "+----------+---+------+\n",
      "|department| id|salary|\n",
      "+----------+---+------+\n",
      "|        HR|  1| 60000|\n",
      "|        HR|  2| 55000|\n",
      "|        HR|  3| 58000|\n",
      "|        IT|  4| 70000|\n",
      "|        IT|  5| 72000|\n",
      "|        IT|  6| 68000|\n",
      "|     Sales|  7| 75000|\n",
      "|     Sales|  8| 78000|\n",
      "|     Sales|  9| 77000|\n",
      "+----------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create DataFrames\n",
    "employee_data = [\n",
    "    (1, \"John\"), (2, \"Alice\"), (3, \"Bob\"), (4, \"Emily\"),\n",
    "    (5, \"David\"), (6, \"Sarah\"), (7, \"Michael\"), (8, \"Lisa\"),\n",
    "    (9, \"William\")\n",
    "]\n",
    "employees = spark.createDataFrame(employee_data, [\"id\", \"name\"])\n",
    "\n",
    "salary_data = [\n",
    "    (\"HR\", 1, 60000), (\"HR\", 2, 55000), (\"HR\", 3, 58000),\n",
    "    (\"IT\", 4, 70000), (\"IT\", 5, 72000), (\"IT\", 6, 68000),\n",
    "    (\"Sales\", 7, 75000), (\"Sales\", 8, 78000), (\"Sales\", 9, 77000)\n",
    "]\n",
    "salaries = spark.createDataFrame(salary_data, [\"department\", \"id\", \"salary\"])\n",
    "\n",
    "employees.show()\n",
    "\n",
    "salaries.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "id": "871a978f-82f9-4388-afae-77d4b0e7297b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register as temporary views\n",
    "employees.createOrReplaceTempView(\"employees\")\n",
    "salaries.createOrReplaceTempView(\"salaries\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "c0a3c981-b8d4-409c-bd3a-aaff55ee3661",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|   name|\n",
      "+-------+\n",
      "|  Emily|\n",
      "|  David|\n",
      "|Michael|\n",
      "|   Lisa|\n",
      "|William|\n",
      "+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Subquery to find employees with salaries above average\n",
    "result = spark.sql(\"\"\"\n",
    "    SELECT name\n",
    "    FROM employees\n",
    "    WHERE id IN (\n",
    "        SELECT id\n",
    "        FROM salaries\n",
    "        WHERE salary > (SELECT AVG(salary) FROM salaries)\n",
    "    )\n",
    "\"\"\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5237a76b-d747-4096-a013-41cc913cd9c0",
   "metadata": {},
   "source": [
    "### Window Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "id": "afd58076-cada-4552-a676-7919cb17ba2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql import functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "5669387e-aa09-4830-a4c4-c32056e38d5f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---+------+-------+\n",
      "|department| id|salary|   name|\n",
      "+----------+---+------+-------+\n",
      "|        HR|  1| 60000|   John|\n",
      "|        HR|  2| 55000|  Alice|\n",
      "|        HR|  3| 58000|    Bob|\n",
      "|        IT|  4| 70000|  Emily|\n",
      "|        IT|  5| 72000|  David|\n",
      "|        IT|  6| 68000|  Sarah|\n",
      "|     Sales|  7| 75000|Michael|\n",
      "|     Sales|  8| 78000|   Lisa|\n",
      "|     Sales|  9| 77000|William|\n",
      "+----------+---+------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "employee_salary = spark.sql(\"\"\"\n",
    "    select  salaries.*, employees.name\n",
    "    from salaries \n",
    "    left join employees on salaries.id = employees.id\n",
    "\"\"\")\n",
    "\n",
    "employee_salary.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "id": "0f2fb287-69a4-4860-a13b-f76646e2f465",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a window specification\n",
    "window_spec = Window.partitionBy(\"department\").orderBy(F.desc(\"salary\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "id": "dcf9ca4b-f1ce-43f1-ac72-bac31c32e34f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---+------+-------+----+\n",
      "|department| id|salary|   name|rank|\n",
      "+----------+---+------+-------+----+\n",
      "|        HR|  1| 60000|   John|   1|\n",
      "|        HR|  3| 58000|    Bob|   2|\n",
      "|        HR|  2| 55000|  Alice|   3|\n",
      "|        IT|  5| 72000|  David|   1|\n",
      "|        IT|  4| 70000|  Emily|   2|\n",
      "|        IT|  6| 68000|  Sarah|   3|\n",
      "|     Sales|  8| 78000|   Lisa|   1|\n",
      "|     Sales|  9| 77000|William|   2|\n",
      "|     Sales|  7| 75000|Michael|   3|\n",
      "+----------+---+------+-------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Calculate the rank of employees within each department based on salary\n",
    "employee_salary.withColumn(\"rank\", F.rank().over(window_spec)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "7d85a9f5-3da2-4ef7-b201-7bb43ccc6bc2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop the SparkSession\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fea9d104-e4b3-4be0-9fae-43fdbc391691",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}