{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# PySpark Transformations — 20 Exhaustive Practice Problems\n", "\n", "**Coverage:** select, filter, withColumn, groupBy, agg, join, window functions, pivot, union, explode, UDFs, orderBy, distinct, dropDuplicates, cast, when/otherwise, null handling, repartition, cache, and more.\n", "\n", "> Dummy data is pre-loaded. Write your transformations in the `# YOUR CODE HERE` cells.\n", "> Solutions are provided at the bottom of this notebook." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from pyspark.sql import SparkSession, functions as F\n", "from pyspark.sql.window import Window\n", "from pyspark.sql.types import (\n", " DateType,\n", " DoubleType,\n", " IntegerType,\n", " StringType,\n", " StructField,\n", " StructType,\n", " TimestampType,\n", ")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Setup — Run This First" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "spark = (\n", " SparkSession.builder\n", " .appName(\"SparkPractice\")\n", " .master(\"local[*]\")\n", " .getOrCreate()\n", ")\n", "\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "print(f\"Spark Ready! Version: {spark.version}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Dummy Data Setup" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ─── employees DataFrame ───────────────────────────────────────────────────\n", "employees_data = [\n", " (1, \"Alice\", \"Engineering\", \"India\", 95000.0, 5, \"2019-03-15\"),\n", " (2, \"Bob\", \"Marketing\", \"US\", 72000.0, 3, \"2021-06-01\"),\n", " (3, \"Charlie\", \"Engineering\", \"US\", 110000.0, 8, \"2016-01-20\"),\n", " (4, \"Diana\", \"HR\", \"India\", 60000.0, 2, \"2022-09-10\"),\n", " (5, \"Eve\", \"Engineering\", \"Germany\", 120000.0, 10, \"2014-07-04\"),\n", " (6, \"Frank\", \"Marketing\", \"India\", 68000.0, 4, \"2020-11-30\"),\n", " (7, \"Grace\", \"HR\", \"US\", 58000.0, 1, \"2023-02-14\"),\n", " (8, \"Hank\", \"Engineering\", \"Germany\", 98000.0, 6, \"2018-05-22\"),\n", " (9, \"Ivy\", \"Marketing\", \"Germany\", 75000.0, 5, \"2019-08-09\"),\n", " (10, \"Jack\", \"HR\", \"US\", None, 3, \"2021-03-17\"),\n", "]\n", "employeesDF = spark.createDataFrame(\n", " employees_data,\n", " [\"emp_id\", \"name\", \"department\", \"country\", \"salary\", \"years_exp\", \"join_date\"],\n", ")\n", "\n", "# ─── sales DataFrame ───────────────────────────────────────────────────────\n", "sales_data = [\n", " (101, 1, \"Laptop\", \"Electronics\", 1200.0, 2, \"2024-01-05\"),\n", " (102, 2, \"Phone\", \"Electronics\", 800.0, 1, \"2024-01-07\"),\n", " (103, 1, \"Desk\", \"Furniture\", 450.0, 3, \"2024-01-10\"),\n", " (104, 3, \"Laptop\", \"Electronics\", 1200.0, 1, \"2024-01-12\"),\n", " (105, 4, \"Chair\", \"Furniture\", 250.0, 4, \"2024-02-01\"),\n", " (106, 5, \"Laptop\", \"Electronics\", 1200.0, 5, \"2024-02-03\"),\n", " (107, 2, \"Monitor\", \"Electronics\", 600.0, 2, \"2024-02-15\"),\n", " (108, 6, \"Keyboard\", \"Electronics\", 80.0, 10, \"2024-03-01\"),\n", " (109, 7, \"Desk\", \"Furniture\", 450.0, 1, \"2024-03-05\"),\n", " (110, 1, \"Phone\", \"Electronics\", 800.0, 1, \"2024-03-10\"),\n", " (111, 8, \"Chair\", \"Furniture\", 250.0, 6, \"2024-03-15\"),\n", " (112, 9, \"Monitor\", \"Electronics\", 600.0, 3, \"2024-04-01\"),\n", "]\n", "salesDF = spark.createDataFrame(\n", " sales_data,\n", " [\"sale_id\", \"emp_id\", \"product\", \"category\", \"price\", \"quantity\", \"sale_date\"],\n", ")\n", "\n", "# ─── departments DataFrame ────────────────────────────────────────────────\n", "departments_data = [\n", " (\"Engineering\", \"Alice Johnson\", 500000.0),\n", " (\"Marketing\", \"Bob Williams\", 300000.0),\n", " (\"HR\", \"Carol Smith\", 150000.0),\n", " (\"Finance\", \"David Brown\", 200000.0), # No employees in this dept\n", "]\n", "departmentsDF = spark.createDataFrame(\n", " departments_data, [\"dept_name\", \"dept_head\", \"budget\"]\n", ")\n", "\n", "# ─── skills DataFrame (array column) ─────────────────────────────────────\n", "skills_data = [\n", " (1, [\"Scala\", \"Spark\", \"SQL\"]),\n", " (2, [\"Marketing\", \"SEO\"]),\n", " (3, [\"Java\", \"Scala\", \"Kafka\", \"Spark\"]),\n", " (4, [\"HR\", \"Recruiting\"]),\n", " (5, [\"Scala\", \"Python\", \"ML\", \"Spark\"]),\n", "]\n", "skillsDF = spark.createDataFrame(skills_data, [\"emp_id\", \"skills\"])\n", "\n", "employeesDF.createOrReplaceTempView(\"employees\")\n", "salesDF.createOrReplaceTempView(\"sales\")\n", "departmentsDF.createOrReplaceTempView(\"departments\")\n", "skillsDF.createOrReplaceTempView(\"skills\")\n", "\n", "print(\"All DataFrames Ready!\")\n", "print(\"\\n--- Employees ---\")\n", "employeesDF.show()\n", "print(\"\\n--- Sales ---\")\n", "salesDF.show()\n", "print(\"\\n--- Departments ---\")\n", "departmentsDF.show()\n", "print(\"\\n--- Skills ---\")\n", "skillsDF.show(truncate=False)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "# PROBLEMS\n", "\n", "---\n", "## Q1 — select, filter, withColumn\n", "\n", "From `employeesDF`:\n", "1. Select `name`, `department`, `salary`\n", "2. Filter only employees in `Engineering` with salary > 90000\n", "3. Add a new column `salary_inr` = salary × 83 (USD to INR)\n", "4. Show the result" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 1 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+-----------+--------+----------+\n", "| name| department| salary|salary_inr|\n", "+-------+-----------+--------+----------+\n", "| Alice|Engineering| 95000.0| 7885000.0|\n", "|Charlie|Engineering|110000.0| 9130000.0|\n", "| Eve|Engineering|120000.0| 9960000.0|\n", "| Hank|Engineering| 98000.0| 8134000.0|\n", "+-------+-----------+--------+----------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q2 — groupBy + agg (count, avg, max, min, sum)\n", "\n", "From `employeesDF`, for each `department` compute:\n", "- total employees (`count`)\n", "- average salary (`avg`)\n", "- max salary\n", "- min salary\n", "- total salary budget (`sum`)\n", "\n", "Round avg to 2 decimal places. Sort by avg salary descending." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 2 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-----------+---------------+----------+----------+----------+------------+\n", "| department|total_employees|avg_salary|max_salary|min_salary|total_salary|\n", "+-----------+---------------+----------+----------+----------+------------+\n", "|Engineering| 4| 105750.0| 120000.0| 95000.0| 423000.0|\n", "| Marketing| 3| 71666.67| 75000.0| 68000.0| 215000.0|\n", "| HR| 3| 59000.0| 60000.0| 58000.0| 118000.0|\n", "+-----------+---------------+----------+----------+----------+------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q3 — Inner, Left, Right & Anti Join\n", "\n", "1. **Inner Join** `employeesDF` with `salesDF` on `emp_id` → show employee name and their sales\n", "2. **Left Join** `employeesDF` with `salesDF` → find employees who made NO sales (anti-join pattern using left + filter null)\n", "3. **Anti Join** — same as above but use `anti` join type directly\n", "4. **Left Join** `departmentsDF` with `employeesDF` → show departments with no employees" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 3 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Inner Join ===\n", "+-------+-----------+--------+------+--------+----------+\n", "| name| department| product| price|quantity| sale_date|\n", "+-------+-----------+--------+------+--------+----------+\n", "| Alice|Engineering| Laptop|1200.0| 2|2024-01-05|\n", "| Alice|Engineering| Desk| 450.0| 3|2024-01-10|\n", "| Alice|Engineering| Phone| 800.0| 1|2024-03-10|\n", "| Bob| Marketing| Phone| 800.0| 1|2024-01-07|\n", "| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|\n", "|Charlie|Engineering| Laptop|1200.0| 1|2024-01-12|\n", "| Diana| HR| Chair| 250.0| 4|2024-02-01|\n", "| Eve|Engineering| Laptop|1200.0| 5|2024-02-03|\n", "| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|\n", "| Grace| HR| Desk| 450.0| 1|2024-03-05|\n", "| Hank|Engineering| Chair| 250.0| 6|2024-03-15|\n", "| Ivy| Marketing| Monitor| 600.0| 3|2024-04-01|\n", "+-------+-----------+--------+------+--------+----------+\n", "\n", "=== No Sales (Left + null filter) ===\n", "+------+----+----------+\n", "|emp_id|name|department|\n", "+------+----+----------+\n", "| 10|Jack| HR|\n", "+------+----+----------+\n", "\n", "=== Anti Join ===\n", "+------+----+----------+\n", "|emp_id|name|department|\n", "+------+----+----------+\n", "| 10|Jack| HR|\n", "+------+----+----------+\n", "\n", "=== Departments with no Employees ===\n", "+---------+-----------+\n", "|dept_name| dept_head|\n", "+---------+-----------+\n", "| Finance|David Brown|\n", "+---------+-----------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q4 — Window Functions: rank, dense_rank, row_number\n", "\n", "From `employeesDF`:\n", "1. Rank employees by salary within each `department` (highest first)\n", "2. Add columns: `rank`, `dense_rank`, `row_number` — all partitioned by `department`, ordered by `salary` desc\n", "3. Filter to show only the **top 2** employees per department by dense_rank" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 4 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== All Ranks ===\n", "+-------+-----------+--------+----+----------+----------+\n", "| name| department| salary|rank|dense_rank|row_number|\n", "+-------+-----------+--------+----+----------+----------+\n", "| Eve|Engineering|120000.0| 1| 1| 1|\n", "|Charlie|Engineering|110000.0| 2| 2| 2|\n", "| Hank|Engineering| 98000.0| 3| 3| 3|\n", "| Alice|Engineering| 95000.0| 4| 4| 4|\n", "| Diana| HR| 60000.0| 1| 1| 1|\n", "| Grace| HR| 58000.0| 2| 2| 2|\n", "| Jack| HR| NULL| 3| 3| 3|\n", "| Ivy| Marketing| 75000.0| 1| 1| 1|\n", "| Bob| Marketing| 72000.0| 2| 2| 2|\n", "| Frank| Marketing| 68000.0| 3| 3| 3|\n", "+-------+-----------+--------+----+----------+----------+\n", "\n", "=== Top 2 per Department ===\n", "+-------+-----------+--------+----------+\n", "| name| department| salary|dense_rank|\n", "+-------+-----------+--------+----------+\n", "| Eve|Engineering|120000.0| 1|\n", "|Charlie|Engineering|110000.0| 2|\n", "| Diana| HR| 60000.0| 1|\n", "| Grace| HR| 58000.0| 2|\n", "| Ivy| Marketing| 75000.0| 1|\n", "| Bob| Marketing| 72000.0| 2|\n", "+-------+-----------+--------+----------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q5 — Window Functions: lag, lead, running total\n", "\n", "From `salesDF` (cast `sale_date` as Date, order by it per `emp_id`):\n", "1. Add `prev_sale_price` = lag of `price` by 1, partitioned by `emp_id`\n", "2. Add `next_sale_price` = lead of `price` by 1, partitioned by `emp_id`\n", "3. Add `running_revenue` = cumulative sum of `price * quantity` ordered by `sale_date`" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 5 }, "source": [ "### Expected output\n", "\n", "```text\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n", "|emp_id| product| price|quantity| sale_date|prev_sale_price|next_sale_price|running_revenue|\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n", "| 1| Laptop|1200.0| 2|2024-01-05| NULL| 450.0| 2400.0|\n", "| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 3750.0|\n", "| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 4550.0|\n", "| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 800.0|\n", "| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 2000.0|\n", "| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 1200.0|\n", "| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 1000.0|\n", "| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 6000.0|\n", "| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 800.0|\n", "| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 450.0|\n", "| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 1500.0|\n", "| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 1800.0|\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q6 — when / otherwise (conditional column)\n", "\n", "From `employeesDF`, create a new column `salary_band`:\n", "- `\"High\"` if salary >= 100000\n", "- `\"Mid\"` if salary >= 70000\n", "- `\"Low\"` if salary >= 0\n", "- `\"Unknown\"` if salary is null\n", "\n", "Also create `experience_level`:\n", "- `\"Senior\"` if years_exp >= 7\n", "- `\"Mid\"` if years_exp >= 4\n", "- `\"Junior\"` otherwise" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 6 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+--------+-----------+---------+----------------+\n", "| name| salary|salary_band|years_exp|experience_level|\n", "+-------+--------+-----------+---------+----------------+\n", "| Alice| 95000.0| Mid| 5| Mid|\n", "| Bob| 72000.0| Mid| 3| Junior|\n", "|Charlie|110000.0| High| 8| Senior|\n", "| Diana| 60000.0| Low| 2| Junior|\n", "| Eve|120000.0| High| 10| Senior|\n", "| Frank| 68000.0| Low| 4| Mid|\n", "| Grace| 58000.0| Low| 1| Junior|\n", "| Hank| 98000.0| Mid| 6| Mid|\n", "| Ivy| 75000.0| Mid| 5| Mid|\n", "| Jack| NULL| Unknown| 3| Junior|\n", "+-------+--------+-----------+---------+----------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q7 — Null Handling: isNull, isNotNull, fillna, dropna, coalesce\n", "\n", "From `employeesDF`:\n", "1. Count rows where `salary` is null\n", "2. Fill null salaries with the **average salary** of the dataset\n", "3. Use `coalesce` to replace null salary with `60000.0` as fallback\n", "4. Drop rows where **any** column is null\n", "5. Drop rows where salary **and** country are both null (subset dropna)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 7 }, "source": [ "### Expected output\n", "\n", "```text\n", "Null salary count: 1\n", "=== Filled with avg ===\n", "+-------+--------+\n", "| name| salary|\n", "+-------+--------+\n", "| Alice| 95000.0|\n", "| Bob| 72000.0|\n", "|Charlie|110000.0|\n", "| Diana| 60000.0|\n", "| Eve|120000.0|\n", "| Frank| 68000.0|\n", "| Grace| 58000.0|\n", "| Hank| 98000.0|\n", "| Ivy| 75000.0|\n", "| Jack| 84000.0|\n", "+-------+--------+\n", "\n", "=== Coalesce ===\n", "+-------+--------+\n", "| name| salary|\n", "+-------+--------+\n", "| Alice| 95000.0|\n", "| Bob| 72000.0|\n", "|Charlie|110000.0|\n", "| Diana| 60000.0|\n", "| Eve|120000.0|\n", "| Frank| 68000.0|\n", "| Grace| 58000.0|\n", "| Hank| 98000.0|\n", "| Ivy| 75000.0|\n", "| Jack| 60000.0|\n", "+-------+--------+\n", "\n", "After na.drop (any): 9 rows\n", "After subset na.drop (all): 10 rows\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q8 — String Functions\n", "\n", "From `employeesDF`:\n", "1. Uppercase `name`, lowercase `department`\n", "2. Concatenate `name` and `department` as `\"Alice (Engineering)\"`\n", "3. Extract first 3 chars of `name` as `name_code`\n", "4. Check if `name` contains the letter `\"a\"` (case-insensitive) → boolean column\n", "5. Replace `\"India\"` with `\"IN\"` and `\"US\"` with `\"USA\"` in the `country` column using regexp_replace" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "vscode": { "languageId": "plaintext" } }, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 8 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+----------+-----------+--------------------+---------+-----+------------+\n", "| name|name_upper| dept_lower| name_dept|name_code|has_a|country_code|\n", "+-------+----------+-----------+--------------------+---------+-----+------------+\n", "| Alice| ALICE|engineering| Alice (Engineering)| Ali| true| IN|\n", "| Bob| BOB| marketing| Bob (Marketing)| Bob|false| USA|\n", "|Charlie| CHARLIE|engineering|Charlie (Engineer...| Cha| true| USA|\n", "| Diana| DIANA| hr| Diana (HR)| Dia| true| IN|\n", "| Eve| EVE|engineering| Eve (Engineering)| Eve|false| Germany|\n", "| Frank| FRANK| marketing| Frank (Marketing)| Fra| true| IN|\n", "| Grace| GRACE| hr| Grace (HR)| Gra| true| USA|\n", "| Hank| HANK|engineering| Hank (Engineering)| Han| true| Germany|\n", "| Ivy| IVY| marketing| Ivy (Marketing)| Ivy|false| Germany|\n", "| Jack| JACK| hr| Jack (HR)| Jac| true| USA|\n", "+-------+----------+-----------+--------------------+---------+-----+------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q9 — Date Functions\n", "\n", "From `employeesDF` (treat `join_date` as a date):\n", "1. Cast `join_date` to `DateType`\n", "2. Extract `year`, `month`, `day` as separate columns\n", "3. Calculate `tenure_days` = today's date minus `join_date`\n", "4. Calculate `tenure_years` (rounded to 1 decimal)\n", "5. Add 30 days to `join_date` as `probation_end`" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 9 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n", "| name| join_date|join_year|join_month|join_day|tenure_days|tenure_years|probation_end|\n", "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n", "| Alice|2019-03-15| 2019| 3| 15| 2764| 7.6| 2019-04-14|\n", "| Bob|2021-06-01| 2021| 6| 1| 1955| 5.4| 2021-07-01|\n", "|Charlie|2016-01-20| 2016| 1| 20| 3914| 10.7| 2016-02-19|\n", "| Diana|2022-09-10| 2022| 9| 10| 1489| 4.1| 2022-10-10|\n", "| Eve|2014-07-04| 2014| 7| 4| 4479| 12.3| 2014-08-03|\n", "| Frank|2020-11-30| 2020| 11| 30| 2138| 5.9| 2020-12-30|\n", "| Grace|2023-02-14| 2023| 2| 14| 1332| 3.6| 2023-03-16|\n", "| Hank|2018-05-22| 2018| 5| 22| 3061| 8.4| 2018-06-21|\n", "| Ivy|2019-08-09| 2019| 8| 9| 2617| 7.2| 2019-09-08|\n", "| Jack|2021-03-17| 2021| 3| 17| 2031| 5.6| 2021-04-16|\n", "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q10 — explode + array functions\n", "\n", "From `skillsDF`:\n", "1. `explode` the `skills` array → one row per skill per employee\n", "2. Find how many skills each employee has (`size` of array)\n", "3. Find all employees who have `\"Spark\"` in their skills (`array_contains`)\n", "4. Join with `employeesDF` to show employee name alongside their individual skills\n", "5. `collect_list` — after explode and join, re-aggregate skills back per employee" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 10 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Exploded ===\n", "+------+----------+\n", "|emp_id| skill|\n", "+------+----------+\n", "| 1| Scala|\n", "| 1| Spark|\n", "| 1| SQL|\n", "| 2| Marketing|\n", "| 2| SEO|\n", "| 3| Java|\n", "| 3| Scala|\n", "| 3| Kafka|\n", "| 3| Spark|\n", "| 4| HR|\n", "| 4|Recruiting|\n", "| 5| Scala|\n", "| 5| Python|\n", "| 5| ML|\n", "| 5| Spark|\n", "+------+----------+\n", "\n", "=== Skill Count ===\n", "+------+--------------------+-----------+\n", "|emp_id| skills|skill_count|\n", "+------+--------------------+-----------+\n", "| 1| [Scala, Spark, SQL]| 3|\n", "| 2| [Marketing, SEO]| 2|\n", "| 3|[Java, Scala, Kaf...| 4|\n", "| 4| [HR, Recruiting]| 2|\n", "| 5|[Scala, Python, M...| 4|\n", "+------+--------------------+-----------+\n", "\n", "=== Spark Developers ===\n", "+------+--------------------+\n", "|emp_id| skills|\n", "+------+--------------------+\n", "| 1| [Scala, Spark, SQL]|\n", "| 3|[Java, Scala, Kaf...|\n", "| 5|[Scala, Python, M...|\n", "+------+--------------------+\n", "\n", "=== Name + Skills ===\n", "+------+-------+----------+\n", "|emp_id| name| skill|\n", "+------+-------+----------+\n", "| 1| Alice| Scala|\n", "| 1| Alice| Spark|\n", "| 1| Alice| SQL|\n", "| 2| Bob| Marketing|\n", "| 2| Bob| SEO|\n", "| 3|Charlie| Java|\n", "| 3|Charlie| Scala|\n", "| 3|Charlie| Kafka|\n", "| 3|Charlie| Spark|\n", "| 4| Diana| HR|\n", "| 4| Diana|Recruiting|\n", "| 5| Eve| Scala|\n", "| 5| Eve| Python|\n", "| 5| Eve| ML|\n", "| 5| Eve| Spark|\n", "+------+-------+----------+\n", "\n", "=== Re-aggregated ===\n", "+------+-------+---------------------------+\n", "|emp_id|name |skills_list |\n", "+------+-------+---------------------------+\n", "|1 |Alice |[Scala, Spark, SQL] |\n", "|2 |Bob |[Marketing, SEO] |\n", "|3 |Charlie|[Java, Scala, Kafka, Spark]|\n", "|4 |Diana |[HR, Recruiting] |\n", "|5 |Eve |[Scala, Python, ML, Spark] |\n", "+------+-------+---------------------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q11 — Pivot Table\n", "\n", "From `salesDF`:\n", "1. Pivot on `category` (columns: `Electronics`, `Furniture`)\n", "2. Rows = `emp_id`\n", "3. Values = total revenue (`sum of price * quantity`)\n", "4. Fill nulls with 0\n", "5. Show the pivoted result" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 11 }, "source": [ "### Expected output\n", "\n", "```text\n", "+------+-----------+---------+\n", "|emp_id|Electronics|Furniture|\n", "+------+-----------+---------+\n", "| 7| 0.0| 450.0|\n", "| 6| 800.0| 0.0|\n", "| 9| 1800.0| 0.0|\n", "| 5| 6000.0| 0.0|\n", "| 1| 3200.0| 1350.0|\n", "| 3| 1200.0| 0.0|\n", "| 8| 0.0| 1500.0|\n", "| 2| 2000.0| 0.0|\n", "| 4| 0.0| 1000.0|\n", "+------+-----------+---------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q12 — Union, UnionByName, Intersect, Except\n", "\n", "1. Create two subsets from `employeesDF`: `engDF` (Engineering) and `mktDF` (Marketing)\n", "2. `union` them and show result\n", "3. `unionByName` — demonstrate with a reordered `mktDF`\n", "4. Create `highSalaryDF` (salary > 80000) and `seniorDF` (years_exp >= 5)\n", "5. Use `intersect` → employees both high-salary AND senior\n", "6. Use `except` → high-salary employees who are NOT senior" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 12 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Union ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|\n", "| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== unionByName ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|\n", "| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== Intersect: High Salary AND Senior ===\n", "+-------+--------+---------+\n", "| name| salary|years_exp|\n", "+-------+--------+---------+\n", "| Eve|120000.0| 10|\n", "| Hank| 98000.0| 6|\n", "|Charlie|110000.0| 8|\n", "| Alice| 95000.0| 5|\n", "+-------+--------+---------+\n", "\n", "=== Except: High Salary but NOT Senior ===\n", "+----+------+---------+\n", "|name|salary|years_exp|\n", "+----+------+---------+\n", "+----+------+---------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q13 — UDF (User Defined Function)\n", "\n", "1. Write a UDF `gradeEmployee` that takes `salary` and returns:\n", " - `\"A\"` if >= 100000, `\"B\"` if >= 75000, `\"C\"` if >= 50000, `\"D\"` otherwise\n", "2. Register and apply it to `employeesDF` as a new column `grade`\n", "3. Write a second UDF `maskName` that returns first char + `\"***\"` (e.g., `\"Alice\"` → `\"A***\"`)\n", "4. Apply `maskName` to the `name` column" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 13 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+-----------+--------+-----+\n", "| name|masked_name| salary|grade|\n", "+-------+-----------+--------+-----+\n", "| Alice| A***| 95000.0| B|\n", "| Bob| B***| 72000.0| C|\n", "|Charlie| C***|110000.0| A|\n", "| Diana| D***| 60000.0| C|\n", "| Eve| E***|120000.0| A|\n", "| Frank| F***| 68000.0| C|\n", "| Grace| G***| 58000.0| C|\n", "| Hank| H***| 98000.0| B|\n", "| Ivy| I***| 75000.0| B|\n", "| Jack| J***| NULL| D|\n", "+-------+-----------+--------+-----+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q14 — distinct, dropDuplicates\n", "\n", "1. From `salesDF`, find all unique `product` names using `distinct`\n", "2. Create a DataFrame with intentional duplicates by union-ing `salesDF` with itself\n", "3. Use `dropDuplicates()` to remove full row duplicates\n", "4. Use `dropDuplicates(Seq(\"emp_id\", \"product\"))` to keep only one sale per employee-product combo\n", "5. Confirm counts at each step" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 14 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Distinct Products ===\n", "+--------+\n", "| product|\n", "+--------+\n", "| Laptop|\n", "| Desk|\n", "| Phone|\n", "| Chair|\n", "| Monitor|\n", "|Keyboard|\n", "+--------+\n", "\n", "With duplicates: 24 rows\n", "After dropDuplicates(): 12 rows\n", "After dropDuplicates(emp_id, product): 12 rows\n", "+-------+------+--------+-----------+------+--------+----------+\n", "|sale_id|emp_id| product| category| price|quantity| sale_date|\n", "+-------+------+--------+-----------+------+--------+----------+\n", "| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|\n", "| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|\n", "| 110| 1| Phone|Electronics| 800.0| 1|2024-03-10|\n", "| 107| 2| Monitor|Electronics| 600.0| 2|2024-02-15|\n", "| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|\n", "| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|\n", "| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|\n", "| 106| 5| Laptop|Electronics|1200.0| 5|2024-02-03|\n", "| 108| 6|Keyboard|Electronics| 80.0| 10|2024-03-01|\n", "| 109| 7| Desk| Furniture| 450.0| 1|2024-03-05|\n", "| 111| 8| Chair| Furniture| 250.0| 6|2024-03-15|\n", "| 112| 9| Monitor|Electronics| 600.0| 3|2024-04-01|\n", "+-------+------+--------+-----------+------+--------+----------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q15 — cast, schema manipulation, printSchema\n", "\n", "1. Print the schema of `employeesDF` and `salesDF`\n", "2. Cast `join_date` (String) to `DateType` in `employeesDF`\n", "3. Cast `salary` to `IntegerType`\n", "4. Cast `sale_date` (String) to `TimestampType` in `salesDF`\n", "5. Define a manual schema using `StructType` + `StructField` and re-read the employees data with it" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 15 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Employees Schema ===\n", "root\n", " |-- emp_id: long (nullable = true)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: double (nullable = true)\n", " |-- years_exp: long (nullable = true)\n", " |-- join_date: string (nullable = true)\n", "\n", "=== Sales Schema ===\n", "root\n", " |-- sale_id: long (nullable = true)\n", " |-- emp_id: long (nullable = true)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = true)\n", " |-- quantity: long (nullable = true)\n", " |-- sale_date: string (nullable = true)\n", "\n", "=== Casted Employees Schema ===\n", "root\n", " |-- emp_id: long (nullable = true)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: integer (nullable = true)\n", " |-- years_exp: long (nullable = true)\n", " |-- join_date: date (nullable = true)\n", "\n", "root\n", " |-- sale_id: long (nullable = true)\n", " |-- emp_id: long (nullable = true)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = true)\n", " |-- quantity: long (nullable = true)\n", " |-- sale_date: timestamp (nullable = true)\n", "\n", "=== Manual Schema ===\n", "root\n", " |-- emp_id: integer (nullable = false)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: double (nullable = true)\n", " |-- years_exp: integer (nullable = true)\n", " |-- join_date: string (nullable = true)\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q16 — orderBy, sort, limit, sample\n", "\n", "1. Sort `employeesDF` by `salary` descending, then by `name` ascending\n", "2. Get the **top 3** highest paid employees\n", "3. Get the **bottom 3** lowest paid employees (handle nulls last)\n", "4. Take a 50% random sample (with seed=42, without replacement)\n", "5. Sort `salesDF` by `sale_date` ascending and show the first 5 records" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 16 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Top 3 Highest Paid ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== Bottom 3 Lowest Paid ===\n", "+------+-----+----------+-------+-------+---------+----------+\n", "|emp_id| name|department|country| salary|years_exp| join_date|\n", "+------+-----+----------+-------+-------+---------+----------+\n", "| 7|Grace| HR| US|58000.0| 1|2023-02-14|\n", "| 4|Diana| HR| India|60000.0| 2|2022-09-10|\n", "| 6|Frank| Marketing| India|68000.0| 4|2020-11-30|\n", "+------+-----+----------+-------+-------+---------+----------+\n", "\n", "=== 50% Sample ===\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 6|Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "| 10| Jack| HR| US| NULL| 3|2021-03-17|\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "\n", "=== First 5 Sales by Date ===\n", "+-------+------+-------+-----------+------+--------+----------+\n", "|sale_id|emp_id|product| category| price|quantity| sale_date|\n", "+-------+------+-------+-----------+------+--------+----------+\n", "| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|\n", "| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|\n", "| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|\n", "| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|\n", "| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|\n", "+-------+------+-------+-----------+------+--------+----------+\n", "only showing top 5 rows\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q17 — repartition, coalesce, cache, persist\n", "\n", "1. Check the current number of partitions in `employeesDF`\n", "2. `repartition(4)` — repartition to 4 partitions (full shuffle)\n", "3. `coalesce(2)` — reduce to 2 partitions without full shuffle\n", "4. `repartition(col(\"department\"))` — partition by department column\n", "5. `cache()` the `employeesDF` and trigger materialization with an action\n", "6. Unpersist/uncache it after use" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 17 }, "source": [ "### Expected output\n", "\n", "```text\n", "Default partitions: 8\n", "After repartition(4): 4\n", "After coalesce(2): 2\n", "After repartition(department): 1\n", "DataFrame cached\n", "DataFrame unpersisted\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q18 — Advanced Aggregations: countDistinct, approx_count_distinct, collect_list, collect_set, first, last\n", "\n", "From `salesDF`:\n", "1. Count **distinct** products sold per `category`\n", "2. Use `approx_count_distinct` for approximate distinct count\n", "3. `collect_list` of all `product` names per `emp_id`\n", "4. `collect_set` of unique products per `emp_id` (no duplicates)\n", "5. Get the `first` and `last` sale per emp_id (by sale_date)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 18 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== countDistinct ===\n", "+-----------+-----------------+\n", "| category|distinct_products|\n", "+-----------+-----------------+\n", "|Electronics| 4|\n", "| Furniture| 2|\n", "+-----------+-----------------+\n", "\n", "=== approx_count_distinct ===\n", "+-----------+---------------+\n", "| category|approx_distinct|\n", "+-----------+---------------+\n", "|Electronics| 4|\n", "| Furniture| 2|\n", "+-----------+---------------+\n", "\n", "=== collect_list ===\n", "+------+---------------------+\n", "|emp_id|all_products |\n", "+------+---------------------+\n", "|1 |[Laptop, Desk, Phone]|\n", "|2 |[Phone, Monitor] |\n", "|3 |[Laptop] |\n", "|5 |[Laptop] |\n", "|4 |[Chair] |\n", "|7 |[Desk] |\n", "|6 |[Keyboard] |\n", "|9 |[Monitor] |\n", "|8 |[Chair] |\n", "+------+---------------------+\n", "\n", "=== collect_set ===\n", "+------+---------------------+\n", "|emp_id|unique_products |\n", "+------+---------------------+\n", "|1 |[Laptop, Phone, Desk]|\n", "|2 |[Phone, Monitor] |\n", "|3 |[Laptop] |\n", "|5 |[Laptop] |\n", "|4 |[Chair] |\n", "|7 |[Desk] |\n", "|6 |[Keyboard] |\n", "|9 |[Monitor] |\n", "|8 |[Chair] |\n", "+------+---------------------+\n", "\n", "=== First & Last Sale ===\n", "+------+-------------+------------+\n", "|emp_id|first_product|last_product|\n", "+------+-------------+------------+\n", "| 1| Laptop| Phone|\n", "| 2| Phone| Monitor|\n", "| 3| Laptop| Laptop|\n", "| 4| Chair| Chair|\n", "| 5| Laptop| Laptop|\n", "| 6| Keyboard| Keyboard|\n", "| 7| Desk| Desk|\n", "| 8| Chair| Chair|\n", "| 9| Monitor| Monitor|\n", "+------+-------------+------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q19 — Complex Pipeline (Multi-step)\n", "\n", "Build a **complete employee performance report** combining multiple transformations:\n", "\n", "1. Join `employeesDF` with `salesDF` on `emp_id` (left join to keep all employees)\n", "2. Calculate `revenue_per_emp` = sum(price × quantity) per employee\n", "3. Fill null revenue (employees with no sales) with 0\n", "4. Join with `departmentsDF` on department = dept_name\n", "5. Add `salary_pct_of_budget` = (salary / budget) × 100\n", "6. Add window-based `dept_salary_rank` = rank within department by salary\n", "7. Add `perf_label` using when/otherwise: `\"Star\"` if revenue > 3000, `\"Good\"` if > 1000, `\"Low\"` otherwise\n", "8. Select: name, department, salary, revenue_per_emp, salary_pct_of_budget, dept_salary_rank, perf_label\n", "9. Sort by department, then dept_salary_rank" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 19 }, "source": [ "### Expected output\n", "\n", "```text\n", "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n", "| name| department| salary|revenue_per_emp|salary_pct_of_budget|dept_salary_rank|perf_label|\n", "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n", "| Eve|Engineering|120000.0| 6000.0| 24.0| 1| Star|\n", "|Charlie|Engineering|110000.0| 1200.0| 22.0| 2| Good|\n", "| Hank|Engineering| 98000.0| 1500.0| 19.6| 3| Good|\n", "| Alice|Engineering| 95000.0| 4550.0| 19.0| 4| Star|\n", "| Diana| HR| 60000.0| 1000.0| 40.0| 1| Low|\n", "| Grace| HR| 58000.0| 450.0| 38.67| 2| Low|\n", "| Jack| HR| NULL| 0.0| NULL| 3| Low|\n", "| Ivy| Marketing| 75000.0| 1800.0| 25.0| 1| Good|\n", "| Bob| Marketing| 72000.0| 2000.0| 24.0| 2| Good|\n", "| Frank| Marketing| 68000.0| 800.0| 22.67| 3| Low|\n", "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q20 — Spark SQL + createOrReplaceTempView\n", "\n", "Use **Spark SQL** (not DataFrame API) to:\n", "1. Find the department with the highest average salary\n", "2. Find employees who earn more than their department's average salary\n", "3. Write a SQL query to get the top 2 earners per department using `ROW_NUMBER() OVER (PARTITION BY ...)`\n", "4. Join all three tables (employees, sales, departments) in one SQL query to compute total revenue per department\n", "5. Use a **CTE** (WITH clause) in SQL to find departments where total salary > 50% of budget" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# YOUR CODE HERE" ] }, { "cell_type": "markdown", "metadata": { "expected_output_for": 20 }, "source": [ "### Expected output\n", "\n", "```text\n", "=== Highest Avg Salary Dept ===\n", "+-----------+----------+\n", "| department|avg_salary|\n", "+-----------+----------+\n", "|Engineering| 105750.0|\n", "+-----------+----------+\n", "\n", "=== Above Dept Average ===\n", "+-------+-----------+--------+--------+\n", "| name| department| salary|dept_avg|\n", "+-------+-----------+--------+--------+\n", "| Bob| Marketing| 72000.0|71666.67|\n", "|Charlie|Engineering|110000.0|105750.0|\n", "| Eve|Engineering|120000.0|105750.0|\n", "| Diana| HR| 60000.0| 59000.0|\n", "| Ivy| Marketing| 75000.0|71666.67|\n", "+-------+-----------+--------+--------+\n", "\n", "=== Top 2 per Dept (SQL Window) ===\n", "+-------+-----------+--------+---+\n", "| name| department| salary| rn|\n", "+-------+-----------+--------+---+\n", "| Eve|Engineering|120000.0| 1|\n", "|Charlie|Engineering|110000.0| 2|\n", "| Diana| HR| 60000.0| 1|\n", "| Grace| HR| 58000.0| 2|\n", "| Ivy| Marketing| 75000.0| 1|\n", "| Bob| Marketing| 72000.0| 2|\n", "+-------+-----------+--------+---+\n", "\n", "=== Revenue per Dept ===\n", "+-----------+-------------+-------------+\n", "| department| dept_head|total_revenue|\n", "+-----------+-------------+-------------+\n", "|Engineering|Alice Johnson| 13250.0|\n", "| Marketing| Bob Williams| 4600.0|\n", "| HR| Carol Smith| 1450.0|\n", "+-----------+-------------+-------------+\n", "\n", "=== Salary > 50% of Budget (CTE) ===\n", "+-----------+------------+--------+-------------+\n", "| department|total_salary| budget|pct_of_budget|\n", "+-----------+------------+--------+-------------+\n", "|Engineering| 423000.0|500000.0| 84.6|\n", "| HR| 118000.0|150000.0| 78.7|\n", "| Marketing| 215000.0|300000.0| 71.7|\n", "+-----------+------------+--------+-------------+\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "---\n", "# SOLUTIONS\n", "\n", "---\n", "## Solution 1 — select, filter, withColumn" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+-----------+--------+----------+\n", "| name| department| salary|salary_inr|\n", "+-------+-----------+--------+----------+\n", "| Alice|Engineering| 95000.0| 7885000.0|\n", "|Charlie|Engineering|110000.0| 9130000.0|\n", "| Eve|Engineering|120000.0| 9960000.0|\n", "| Hank|Engineering| 98000.0| 8134000.0|\n", "+-------+-----------+--------+----------+\n" ] } ], "source": [ "q1 = (\n", " employeesDF\n", " .select(\"name\", \"department\", \"salary\")\n", " .filter((F.col(\"department\") == \"Engineering\") & (F.col(\"salary\") > 90000))\n", " .withColumn(\"salary_inr\", F.col(\"salary\") * 83)\n", ")\n", "\n", "q1.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 2 — groupBy + agg" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----------+---------------+----------+----------+----------+------------+\n", "| department|total_employees|avg_salary|max_salary|min_salary|total_salary|\n", "+-----------+---------------+----------+----------+----------+------------+\n", "|Engineering| 4| 105750.0| 120000.0| 95000.0| 423000.0|\n", "| Marketing| 3| 71666.67| 75000.0| 68000.0| 215000.0|\n", "| HR| 3| 59000.0| 60000.0| 58000.0| 118000.0|\n", "+-----------+---------------+----------+----------+----------+------------+\n" ] } ], "source": [ "q2 = (\n", " employeesDF\n", " .groupBy(\"department\")\n", " .agg(\n", " F.count(\"emp_id\").alias(\"total_employees\"),\n", " F.round(F.avg(\"salary\"), 2).alias(\"avg_salary\"),\n", " F.max(\"salary\").alias(\"max_salary\"),\n", " F.min(\"salary\").alias(\"min_salary\"),\n", " F.sum(\"salary\").alias(\"total_salary\"),\n", " )\n", " .orderBy(F.col(\"avg_salary\").desc())\n", ")\n", "\n", "q2.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 3 — Joins" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Inner Join ===\n", "+-------+-----------+--------+------+--------+----------+\n", "| name| department| product| price|quantity| sale_date|\n", "+-------+-----------+--------+------+--------+----------+\n", "| Alice|Engineering| Laptop|1200.0| 2|2024-01-05|\n", "| Alice|Engineering| Desk| 450.0| 3|2024-01-10|\n", "| Alice|Engineering| Phone| 800.0| 1|2024-03-10|\n", "| Bob| Marketing| Phone| 800.0| 1|2024-01-07|\n", "| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|\n", "|Charlie|Engineering| Laptop|1200.0| 1|2024-01-12|\n", "| Diana| HR| Chair| 250.0| 4|2024-02-01|\n", "| Eve|Engineering| Laptop|1200.0| 5|2024-02-03|\n", "| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|\n", "| Grace| HR| Desk| 450.0| 1|2024-03-05|\n", "| Hank|Engineering| Chair| 250.0| 6|2024-03-15|\n", "| Ivy| Marketing| Monitor| 600.0| 3|2024-04-01|\n", "+-------+-----------+--------+------+--------+----------+\n", "\n", "=== No Sales (Left + null filter) ===\n", "+------+----+----------+\n", "|emp_id|name|department|\n", "+------+----+----------+\n", "| 10|Jack| HR|\n", "+------+----+----------+\n", "\n", "=== Anti Join ===\n", "+------+----+----------+\n", "|emp_id|name|department|\n", "+------+----+----------+\n", "| 10|Jack| HR|\n", "+------+----+----------+\n", "\n", "=== Departments with no Employees ===\n", "+---------+-----------+\n", "|dept_name| dept_head|\n", "+---------+-----------+\n", "| Finance|David Brown|\n", "+---------+-----------+\n" ] } ], "source": [ "# 1. Inner Join\n", "inner_join = (\n", " employeesDF\n", " .join(salesDF, \"emp_id\", \"inner\")\n", " .select(\"name\", \"department\", \"product\", \"price\", \"quantity\", \"sale_date\")\n", ")\n", "print(\"=== Inner Join ===\")\n", "inner_join.show()\n", "\n", "# 2. Left Join + null filter (employees with no sales)\n", "no_sales = (\n", " employeesDF\n", " .join(salesDF, [\"emp_id\"], \"left\")\n", " .filter(F.col(\"sale_id\").isNull())\n", " .select(\"emp_id\", \"name\", \"department\")\n", ")\n", "print(\"=== No Sales (Left + null filter) ===\")\n", "no_sales.show()\n", "\n", "# 3. Anti Join\n", "anti_join = (\n", " employeesDF\n", " .join(salesDF, [\"emp_id\"], \"left_anti\")\n", " .select(\"emp_id\", \"name\", \"department\")\n", ")\n", "print(\"=== Anti Join ===\")\n", "anti_join.show()\n", "\n", "# 4. Departments with no employees\n", "dept_no_emp = (\n", " departmentsDF.alias(\"d\")\n", " .join(\n", " employeesDF.alias(\"e\"),\n", " F.col(\"d.dept_name\") == F.col(\"e.department\"),\n", " \"left\",\n", " )\n", " .filter(F.col(\"e.emp_id\").isNull())\n", " .select(F.col(\"d.dept_name\"), F.col(\"d.dept_head\"))\n", ")\n", "print(\"=== Departments with no Employees ===\")\n", "dept_no_emp.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 4 — rank, dense_rank, row_number" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== All Ranks ===\n", "+-------+-----------+--------+----+----------+----------+\n", "| name| department| salary|rank|dense_rank|row_number|\n", "+-------+-----------+--------+----+----------+----------+\n", "| Eve|Engineering|120000.0| 1| 1| 1|\n", "|Charlie|Engineering|110000.0| 2| 2| 2|\n", "| Hank|Engineering| 98000.0| 3| 3| 3|\n", "| Alice|Engineering| 95000.0| 4| 4| 4|\n", "| Diana| HR| 60000.0| 1| 1| 1|\n", "| Grace| HR| 58000.0| 2| 2| 2|\n", "| Jack| HR| NULL| 3| 3| 3|\n", "| Ivy| Marketing| 75000.0| 1| 1| 1|\n", "| Bob| Marketing| 72000.0| 2| 2| 2|\n", "| Frank| Marketing| 68000.0| 3| 3| 3|\n", "+-------+-----------+--------+----+----------+----------+\n", "\n", "=== Top 2 per Department ===\n", "+-------+-----------+--------+----------+\n", "| name| department| salary|dense_rank|\n", "+-------+-----------+--------+----------+\n", "| Eve|Engineering|120000.0| 1|\n", "|Charlie|Engineering|110000.0| 2|\n", "| Diana| HR| 60000.0| 1|\n", "| Grace| HR| 58000.0| 2|\n", "| Ivy| Marketing| 75000.0| 1|\n", "| Bob| Marketing| 72000.0| 2|\n", "+-------+-----------+--------+----------+\n" ] } ], "source": [ "window_spec = Window.partitionBy(\"department\").orderBy(F.col(\"salary\").desc())\n", "\n", "ranked = (\n", " employeesDF\n", " .withColumn(\"rank\", F.rank().over(window_spec))\n", " .withColumn(\"dense_rank\", F.dense_rank().over(window_spec))\n", " .withColumn(\"row_number\", F.row_number().over(window_spec))\n", ")\n", "\n", "print(\"=== All Ranks ===\")\n", "ranked.select(\n", " \"name\", \"department\", \"salary\", \"rank\", \"dense_rank\", \"row_number\"\n", ").show()\n", "\n", "print(\"=== Top 2 per Department ===\")\n", "(\n", " ranked\n", " .filter(F.col(\"dense_rank\") <= 2)\n", " .select(\"name\", \"department\", \"salary\", \"dense_rank\")\n", " .show()\n", ")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 5 — lag, lead, running total" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------+--------+------+--------+----------+---------------+---------------+---------------+\n", "|emp_id| product| price|quantity| sale_date|prev_sale_price|next_sale_price|running_revenue|\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n", "| 1| Laptop|1200.0| 2|2024-01-05| NULL| 450.0| 2400.0|\n", "| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 3750.0|\n", "| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 4550.0|\n", "| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 800.0|\n", "| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 2000.0|\n", "| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 1200.0|\n", "| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 1000.0|\n", "| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 6000.0|\n", "| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 800.0|\n", "| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 450.0|\n", "| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 1500.0|\n", "| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 1800.0|\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n" ] } ], "source": [ "sales_with_date = salesDF.withColumn(\"sale_date\", F.to_date(\"sale_date\"))\n", "\n", "employee_window = Window.partitionBy(\"emp_id\").orderBy(\"sale_date\")\n", "running_window = employee_window.rowsBetween(\n", " Window.unboundedPreceding, Window.currentRow\n", ")\n", "\n", "q5 = (\n", " sales_with_date\n", " .withColumn(\"prev_sale_price\", F.lag(\"price\", 1).over(employee_window))\n", " .withColumn(\"next_sale_price\", F.lead(\"price\", 1).over(employee_window))\n", " .withColumn(\"revenue\", F.col(\"price\") * F.col(\"quantity\"))\n", " .withColumn(\"running_revenue\", F.sum(\"revenue\").over(running_window))\n", ")\n", "\n", "q5.select(\n", " \"emp_id\", \"product\", \"price\", \"quantity\", \"sale_date\",\n", " \"prev_sale_price\", \"next_sale_price\", \"running_revenue\",\n", ").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 6 — when / otherwise" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+--------+-----------+---------+----------------+\n", "| name| salary|salary_band|years_exp|experience_level|\n", "+-------+--------+-----------+---------+----------------+\n", "| Alice| 95000.0| Mid| 5| Mid|\n", "| Bob| 72000.0| Mid| 3| Junior|\n", "|Charlie|110000.0| High| 8| Senior|\n", "| Diana| 60000.0| Low| 2| Junior|\n", "| Eve|120000.0| High| 10| Senior|\n", "| Frank| 68000.0| Low| 4| Mid|\n", "| Grace| 58000.0| Low| 1| Junior|\n", "| Hank| 98000.0| Mid| 6| Mid|\n", "| Ivy| 75000.0| Mid| 5| Mid|\n", "| Jack| NULL| Unknown| 3| Junior|\n", "+-------+--------+-----------+---------+----------------+\n" ] } ], "source": [ "q6 = (\n", " employeesDF\n", " .withColumn(\n", " \"salary_band\",\n", " F.when(F.col(\"salary\").isNull(), \"Unknown\")\n", " .when(F.col(\"salary\") >= 100000, \"High\")\n", " .when(F.col(\"salary\") >= 70000, \"Mid\")\n", " .otherwise(\"Low\"),\n", " )\n", " .withColumn(\n", " \"experience_level\",\n", " F.when(F.col(\"years_exp\") >= 7, \"Senior\")\n", " .when(F.col(\"years_exp\") >= 4, \"Mid\")\n", " .otherwise(\"Junior\"),\n", " )\n", ")\n", "\n", "q6.select(\n", " \"name\", \"salary\", \"salary_band\", \"years_exp\", \"experience_level\"\n", ").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 7 — Null Handling" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Null salary count: 1\n", "=== Filled with avg ===\n", "+-------+--------+\n", "| name| salary|\n", "+-------+--------+\n", "| Alice| 95000.0|\n", "| Bob| 72000.0|\n", "|Charlie|110000.0|\n", "| Diana| 60000.0|\n", "| Eve|120000.0|\n", "| Frank| 68000.0|\n", "| Grace| 58000.0|\n", "| Hank| 98000.0|\n", "| Ivy| 75000.0|\n", "| Jack| 84000.0|\n", "+-------+--------+\n", "\n", "=== Coalesce ===\n", "+-------+--------+\n", "| name| salary|\n", "+-------+--------+\n", "| Alice| 95000.0|\n", "| Bob| 72000.0|\n", "|Charlie|110000.0|\n", "| Diana| 60000.0|\n", "| Eve|120000.0|\n", "| Frank| 68000.0|\n", "| Grace| 58000.0|\n", "| Hank| 98000.0|\n", "| Ivy| 75000.0|\n", "| Jack| 60000.0|\n", "+-------+--------+\n", "\n", "After na.drop (any): 9 rows\n", "After subset na.drop (all): 10 rows\n" ] } ], "source": [ "# 1. Count null salaries\n", "null_count = employeesDF.filter(F.col(\"salary\").isNull()).count()\n", "print(f\"Null salary count: {null_count}\")\n", "\n", "# 2. Fill null with the average salary\n", "avg_salary = employeesDF.agg(F.avg(\"salary\")).first()[0]\n", "filled = employeesDF.na.fill({\"salary\": avg_salary})\n", "print(\"=== Filled with avg ===\")\n", "filled.select(\"name\", \"salary\").show()\n", "\n", "# 3. coalesce\n", "with_coalesce = employeesDF.withColumn(\n", " \"salary\", F.coalesce(F.col(\"salary\"), F.lit(60000.0))\n", ")\n", "print(\"=== Coalesce ===\")\n", "with_coalesce.select(\"name\", \"salary\").show()\n", "\n", "# 4. Drop rows where any column is null\n", "drop_all = employeesDF.na.drop()\n", "print(f\"After na.drop (any): {drop_all.count()} rows\")\n", "\n", "# 5. Drop rows only when both salary and country are null\n", "drop_subset = employeesDF.na.drop(how=\"all\", subset=[\"salary\", \"country\"])\n", "print(f\"After subset na.drop (all): {drop_subset.count()} rows\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 8 — String Functions" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+----------+-----------+--------------------+---------+-----+------------+\n", "| name|name_upper| dept_lower| name_dept|name_code|has_a|country_code|\n", "+-------+----------+-----------+--------------------+---------+-----+------------+\n", "| Alice| ALICE|engineering| Alice (Engineering)| Ali| true| IN|\n", "| Bob| BOB| marketing| Bob (Marketing)| Bob|false| USA|\n", "|Charlie| CHARLIE|engineering|Charlie (Engineer...| Cha| true| USA|\n", "| Diana| DIANA| hr| Diana (HR)| Dia| true| IN|\n", "| Eve| EVE|engineering| Eve (Engineering)| Eve|false| Germany|\n", "| Frank| FRANK| marketing| Frank (Marketing)| Fra| true| IN|\n", "| Grace| GRACE| hr| Grace (HR)| Gra| true| USA|\n", "| Hank| HANK|engineering| Hank (Engineering)| Han| true| Germany|\n", "| Ivy| IVY| marketing| Ivy (Marketing)| Ivy|false| Germany|\n", "| Jack| JACK| hr| Jack (HR)| Jac| true| USA|\n", "+-------+----------+-----------+--------------------+---------+-----+------------+\n" ] } ], "source": [ "q8 = (\n", " employeesDF\n", " .withColumn(\"name_upper\", F.upper(\"name\"))\n", " .withColumn(\"dept_lower\", F.lower(\"department\"))\n", " .withColumn(\n", " \"name_dept\",\n", " F.concat(F.col(\"name\"), F.lit(\" (\"), F.col(\"department\"), F.lit(\")\")),\n", " )\n", " .withColumn(\"name_code\", F.substring(\"name\", 1, 3))\n", " .withColumn(\"has_a\", F.lower(\"name\").contains(\"a\"))\n", " .withColumn(\n", " \"country_code\",\n", " F.regexp_replace(\n", " F.regexp_replace(F.col(\"country\"), \"India\", \"IN\"), \"^US$\", \"USA\"\n", " ),\n", " )\n", ")\n", "\n", "q8.select(\n", " \"name\", \"name_upper\", \"dept_lower\", \"name_dept\",\n", " \"name_code\", \"has_a\", \"country_code\",\n", ").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 9 — Date Functions" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n", "| name| join_date|join_year|join_month|join_day|tenure_days|tenure_years|probation_end|\n", "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n", "| Alice|2019-03-15| 2019| 3| 15| 2764| 7.6| 2019-04-14|\n", "| Bob|2021-06-01| 2021| 6| 1| 1955| 5.4| 2021-07-01|\n", "|Charlie|2016-01-20| 2016| 1| 20| 3914| 10.7| 2016-02-19|\n", "| Diana|2022-09-10| 2022| 9| 10| 1489| 4.1| 2022-10-10|\n", "| Eve|2014-07-04| 2014| 7| 4| 4479| 12.3| 2014-08-03|\n", "| Frank|2020-11-30| 2020| 11| 30| 2138| 5.9| 2020-12-30|\n", "| Grace|2023-02-14| 2023| 2| 14| 1332| 3.6| 2023-03-16|\n", "| Hank|2018-05-22| 2018| 5| 22| 3061| 8.4| 2018-06-21|\n", "| Ivy|2019-08-09| 2019| 8| 9| 2617| 7.2| 2019-09-08|\n", "| Jack|2021-03-17| 2021| 3| 17| 2031| 5.6| 2021-04-16|\n", "+-------+----------+---------+----------+--------+-----------+------------+-------------+\n" ] } ], "source": [ "q9 = (\n", " employeesDF\n", " .withColumn(\"join_date\", F.to_date(\"join_date\", \"yyyy-MM-dd\"))\n", " .withColumn(\"join_year\", F.year(\"join_date\"))\n", " .withColumn(\"join_month\", F.month(\"join_date\"))\n", " .withColumn(\"join_day\", F.dayofmonth(\"join_date\"))\n", " .withColumn(\"tenure_days\", F.datediff(F.current_date(), F.col(\"join_date\")))\n", " .withColumn(\"tenure_years\", F.round(F.col(\"tenure_days\") / 365.0, 1))\n", " .withColumn(\"probation_end\", F.date_add(F.col(\"join_date\"), 30))\n", ")\n", "\n", "q9.select(\n", " \"name\", \"join_date\", \"join_year\", \"join_month\", \"join_day\",\n", " \"tenure_days\", \"tenure_years\", \"probation_end\",\n", ").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 10 — explode + array functions" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Exploded ===\n", "+------+----------+\n", "|emp_id| skill|\n", "+------+----------+\n", "| 1| Scala|\n", "| 1| Spark|\n", "| 1| SQL|\n", "| 2| Marketing|\n", "| 2| SEO|\n", "| 3| Java|\n", "| 3| Scala|\n", "| 3| Kafka|\n", "| 3| Spark|\n", "| 4| HR|\n", "| 4|Recruiting|\n", "| 5| Scala|\n", "| 5| Python|\n", "| 5| ML|\n", "| 5| Spark|\n", "+------+----------+\n", "\n", "=== Skill Count ===\n", "+------+--------------------+-----------+\n", "|emp_id| skills|skill_count|\n", "+------+--------------------+-----------+\n", "| 1| [Scala, Spark, SQL]| 3|\n", "| 2| [Marketing, SEO]| 2|\n", "| 3|[Java, Scala, Kaf...| 4|\n", "| 4| [HR, Recruiting]| 2|\n", "| 5|[Scala, Python, M...| 4|\n", "+------+--------------------+-----------+\n", "\n", "=== Spark Developers ===\n", "+------+--------------------+\n", "|emp_id| skills|\n", "+------+--------------------+\n", "| 1| [Scala, Spark, SQL]|\n", "| 3|[Java, Scala, Kaf...|\n", "| 5|[Scala, Python, M...|\n", "+------+--------------------+\n", "\n", "=== Name + Skills ===\n", "+------+-------+----------+\n", "|emp_id| name| skill|\n", "+------+-------+----------+\n", "| 1| Alice| Scala|\n", "| 1| Alice| Spark|\n", "| 1| Alice| SQL|\n", "| 2| Bob| Marketing|\n", "| 2| Bob| SEO|\n", "| 3|Charlie| Java|\n", "| 3|Charlie| Scala|\n", "| 3|Charlie| Kafka|\n", "| 3|Charlie| Spark|\n", "| 4| Diana| HR|\n", "| 4| Diana|Recruiting|\n", "| 5| Eve| Scala|\n", "| 5| Eve| Python|\n", "| 5| Eve| ML|\n", "| 5| Eve| Spark|\n", "+------+-------+----------+\n", "\n", "=== Re-aggregated ===\n", "+------+-------+---------------------------+\n", "|emp_id|name |skills_list |\n", "+------+-------+---------------------------+\n", "|1 |Alice |[Scala, Spark, SQL] |\n", "|2 |Bob |[Marketing, SEO] |\n", "|3 |Charlie|[Java, Scala, Kafka, Spark]|\n", "|4 |Diana |[HR, Recruiting] |\n", "|5 |Eve |[Scala, Python, ML, Spark] |\n", "+------+-------+---------------------------+\n" ] } ], "source": [ "# 1. Explode\n", "exploded_df = skillsDF.withColumn(\"skill\", F.explode(\"skills\"))\n", "print(\"=== Exploded ===\")\n", "exploded_df.select(\"emp_id\", \"skill\").show()\n", "\n", "# 2. Array size\n", "with_size = skillsDF.withColumn(\"skill_count\", F.size(\"skills\"))\n", "print(\"=== Skill Count ===\")\n", "with_size.show()\n", "\n", "# 3. array_contains\n", "spark_devs = skillsDF.filter(F.array_contains(\"skills\", \"Spark\"))\n", "print(\"=== Spark Developers ===\")\n", "spark_devs.show()\n", "\n", "# 4. Join with employees\n", "with_names = (\n", " exploded_df\n", " .join(employeesDF, \"emp_id\")\n", " .select(\"emp_id\", \"name\", \"skill\")\n", ")\n", "print(\"=== Name + Skills ===\")\n", "with_names.show()\n", "\n", "# 5. collect_list\n", "reaggregated = with_names.groupBy(\"emp_id\", \"name\").agg(\n", " F.collect_list(\"skill\").alias(\"skills_list\")\n", ")\n", "print(\"=== Re-aggregated ===\")\n", "reaggregated.show(truncate=False)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 11 — Pivot Table" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------+-----------+---------+\n", "|emp_id|Electronics|Furniture|\n", "+------+-----------+---------+\n", "| 7| 0.0| 450.0|\n", "| 6| 800.0| 0.0|\n", "| 9| 1800.0| 0.0|\n", "| 5| 6000.0| 0.0|\n", "| 1| 3200.0| 1350.0|\n", "| 3| 1200.0| 0.0|\n", "| 8| 0.0| 1500.0|\n", "| 2| 2000.0| 0.0|\n", "| 4| 0.0| 1000.0|\n", "+------+-----------+---------+\n" ] } ], "source": [ "q11 = (\n", " salesDF\n", " .withColumn(\"revenue\", F.col(\"price\") * F.col(\"quantity\"))\n", " .groupBy(\"emp_id\")\n", " .pivot(\"category\", [\"Electronics\", \"Furniture\"])\n", " .agg(F.round(F.sum(\"revenue\"), 2))\n", " .na.fill(0)\n", ")\n", "\n", "q11.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 12 — Union, Intersect, Except" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Union ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|\n", "| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== unionByName ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|\n", "| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== Intersect: High Salary AND Senior ===\n", "+-------+--------+---------+\n", "| name| salary|years_exp|\n", "+-------+--------+---------+\n", "| Eve|120000.0| 10|\n", "| Hank| 98000.0| 6|\n", "|Charlie|110000.0| 8|\n", "| Alice| 95000.0| 5|\n", "+-------+--------+---------+\n", "\n", "=== Except: High Salary but NOT Senior ===\n", "+----+------+---------+\n", "|name|salary|years_exp|\n", "+----+------+---------+\n", "+----+------+---------+\n" ] } ], "source": [ "eng_df = employeesDF.filter(F.col(\"department\") == \"Engineering\")\n", "mkt_df = employeesDF.filter(F.col(\"department\") == \"Marketing\")\n", "\n", "# Union\n", "print(\"=== Union ===\")\n", "eng_df.union(mkt_df).show()\n", "\n", "# unionByName with reordered columns\n", "mkt_reordered = mkt_df.select(\n", " \"department\", \"name\", \"emp_id\", \"country\", \"salary\", \"years_exp\", \"join_date\"\n", ")\n", "print(\"=== unionByName ===\")\n", "eng_df.unionByName(mkt_reordered).show()\n", "\n", "# Intersect and set difference\n", "high_salary_df = employeesDF.filter(F.col(\"salary\") > 80000)\n", "senior_df = employeesDF.filter(F.col(\"years_exp\") >= 5)\n", "\n", "print(\"=== Intersect: High Salary AND Senior ===\")\n", "high_salary_df.intersect(senior_df).select(\"name\", \"salary\", \"years_exp\").show()\n", "\n", "print(\"=== Except: High Salary but NOT Senior ===\")\n", "high_salary_df.subtract(senior_df).select(\"name\", \"salary\", \"years_exp\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 13 — UDFs" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+-----------+--------+-----+\n", "| name|masked_name| salary|grade|\n", "+-------+-----------+--------+-----+\n", "| Alice| A***| 95000.0| B|\n", "| Bob| B***| 72000.0| C|\n", "|Charlie| C***|110000.0| A|\n", "| Diana| D***| 60000.0| C|\n", "| Eve| E***|120000.0| A|\n", "| Frank| F***| 68000.0| C|\n", "| Grace| G***| 58000.0| C|\n", "| Hank| H***| 98000.0| B|\n", "| Ivy| I***| 75000.0| B|\n", "| Jack| J***| NULL| D|\n", "+-------+-----------+--------+-----+\n" ] } ], "source": [ "def grade_employee(salary):\n", " if salary is None:\n", " return \"D\"\n", " if salary >= 100000:\n", " return \"A\"\n", " if salary >= 75000:\n", " return \"B\"\n", " if salary >= 50000:\n", " return \"C\"\n", " return \"D\"\n", "\n", "\n", "def mask_name(name):\n", " if not name:\n", " return \"\"\n", " return name[0] + \"***\"\n", "\n", "\n", "grade_employee_udf = F.udf(grade_employee, StringType())\n", "mask_name_udf = F.udf(mask_name, StringType())\n", "\n", "spark.udf.register(\"gradeEmployee\", grade_employee, StringType())\n", "spark.udf.register(\"maskName\", mask_name, StringType())\n", "\n", "q13 = (\n", " employeesDF\n", " .withColumn(\"grade\", grade_employee_udf(\"salary\"))\n", " .withColumn(\"masked_name\", mask_name_udf(\"name\"))\n", ")\n", "\n", "q13.select(\"name\", \"masked_name\", \"salary\", \"grade\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 14 — distinct, dropDuplicates" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Distinct Products ===\n", "+--------+\n", "| product|\n", "+--------+\n", "| Laptop|\n", "| Desk|\n", "| Phone|\n", "| Chair|\n", "| Monitor|\n", "|Keyboard|\n", "+--------+\n", "\n", "With duplicates: 24 rows\n", "After dropDuplicates(): 12 rows\n", "After dropDuplicates(emp_id, product): 12 rows\n", "+-------+------+--------+-----------+------+--------+----------+\n", "|sale_id|emp_id| product| category| price|quantity| sale_date|\n", "+-------+------+--------+-----------+------+--------+----------+\n", "| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|\n", "| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|\n", "| 110| 1| Phone|Electronics| 800.0| 1|2024-03-10|\n", "| 107| 2| Monitor|Electronics| 600.0| 2|2024-02-15|\n", "| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|\n", "| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|\n", "| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|\n", "| 106| 5| Laptop|Electronics|1200.0| 5|2024-02-03|\n", "| 108| 6|Keyboard|Electronics| 80.0| 10|2024-03-01|\n", "| 109| 7| Desk| Furniture| 450.0| 1|2024-03-05|\n", "| 111| 8| Chair| Furniture| 250.0| 6|2024-03-15|\n", "| 112| 9| Monitor|Electronics| 600.0| 3|2024-04-01|\n", "+-------+------+--------+-----------+------+--------+----------+\n" ] } ], "source": [ "# 1. Distinct products\n", "print(\"=== Distinct Products ===\")\n", "salesDF.select(\"product\").distinct().show()\n", "\n", "# 2. Create duplicates\n", "dupe_df = salesDF.union(salesDF)\n", "print(f\"With duplicates: {dupe_df.count()} rows\")\n", "\n", "# 3. Remove full-row duplicates\n", "deduped = dupe_df.dropDuplicates()\n", "print(f\"After dropDuplicates(): {deduped.count()} rows\")\n", "\n", "# 4. Remove duplicates using a column subset\n", "one_per_emp_product = dupe_df.dropDuplicates([\"emp_id\", \"product\"])\n", "print(\n", " f\"After dropDuplicates(emp_id, product): {one_per_emp_product.count()} rows\"\n", ")\n", "one_per_emp_product.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 15 — cast, schema, StructType" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Employees Schema ===\n", "root\n", " |-- emp_id: long (nullable = true)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: double (nullable = true)\n", " |-- years_exp: long (nullable = true)\n", " |-- join_date: string (nullable = true)\n", "\n", "=== Sales Schema ===\n", "root\n", " |-- sale_id: long (nullable = true)\n", " |-- emp_id: long (nullable = true)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = true)\n", " |-- quantity: long (nullable = true)\n", " |-- sale_date: string (nullable = true)\n", "\n", "=== Casted Employees Schema ===\n", "root\n", " |-- emp_id: long (nullable = true)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: integer (nullable = true)\n", " |-- years_exp: long (nullable = true)\n", " |-- join_date: date (nullable = true)\n", "\n", "root\n", " |-- sale_id: long (nullable = true)\n", " |-- emp_id: long (nullable = true)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = true)\n", " |-- quantity: long (nullable = true)\n", " |-- sale_date: timestamp (nullable = true)\n", "\n", "=== Manual Schema ===\n", "root\n", " |-- emp_id: integer (nullable = false)\n", " |-- name: string (nullable = true)\n", " |-- department: string (nullable = true)\n", " |-- country: string (nullable = true)\n", " |-- salary: double (nullable = true)\n", " |-- years_exp: integer (nullable = true)\n", " |-- join_date: string (nullable = true)\n" ] } ], "source": [ "# 1. Print schemas\n", "print(\"=== Employees Schema ===\")\n", "employeesDF.printSchema()\n", "print(\"=== Sales Schema ===\")\n", "salesDF.printSchema()\n", "\n", "# 2 and 3. Cast employee columns\n", "emp_casted = (\n", " employeesDF\n", " .withColumn(\"join_date\", F.col(\"join_date\").cast(DateType()))\n", " .withColumn(\"salary\", F.col(\"salary\").cast(IntegerType()))\n", ")\n", "print(\"=== Casted Employees Schema ===\")\n", "emp_casted.printSchema()\n", "\n", "# 4. Cast sale_date to timestamp\n", "sales_casted = salesDF.withColumn(\n", " \"sale_date\", F.col(\"sale_date\").cast(TimestampType())\n", ")\n", "sales_casted.printSchema()\n", "\n", "# 5. Define and apply a manual schema\n", "manual_schema = StructType([\n", " StructField(\"emp_id\", IntegerType(), nullable=False),\n", " StructField(\"name\", StringType(), nullable=True),\n", " StructField(\"department\", StringType(), nullable=True),\n", " StructField(\"country\", StringType(), nullable=True),\n", " StructField(\"salary\", DoubleType(), nullable=True),\n", " StructField(\"years_exp\", IntegerType(), nullable=True),\n", " StructField(\"join_date\", StringType(), nullable=True),\n", "])\n", "\n", "manual_employees_df = spark.createDataFrame(employees_data, schema=manual_schema)\n", "print(\"=== Manual Schema ===\")\n", "manual_employees_df.printSchema()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 16 — orderBy, limit, sample" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Top 3 Highest Paid ===\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "=== Bottom 3 Lowest Paid ===\n", "+------+-----+----------+-------+-------+---------+----------+\n", "|emp_id| name|department|country| salary|years_exp| join_date|\n", "+------+-----+----------+-------+-------+---------+----------+\n", "| 7|Grace| HR| US|58000.0| 1|2023-02-14|\n", "| 4|Diana| HR| India|60000.0| 2|2022-09-10|\n", "| 6|Frank| Marketing| India|68000.0| 4|2020-11-30|\n", "+------+-----+----------+-------+-------+---------+----------+\n", "\n", "=== 50% Sample ===\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "|emp_id| name| department|country| salary|years_exp| join_date|\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|\n", "| 6|Frank| Marketing| India| 68000.0| 4|2020-11-30|\n", "| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|\n", "| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|\n", "| 10| Jack| HR| US| NULL| 3|2021-03-17|\n", "+------+-----+-----------+-------+--------+---------+----------+\n", "\n", "=== First 5 Sales by Date ===\n", "+-------+------+-------+-----------+------+--------+----------+\n", "|sale_id|emp_id|product| category| price|quantity| sale_date|\n", "+-------+------+-------+-----------+------+--------+----------+\n", "| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|\n", "| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|\n", "| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|\n", "| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|\n", "| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|\n", "+-------+------+-------+-----------+------+--------+----------+\n", "only showing top 5 rows\n" ] } ], "source": [ "# 1 and 2. Sort and get the top 3\n", "print(\"=== Top 3 Highest Paid ===\")\n", "(\n", " employeesDF\n", " .orderBy(F.col(\"salary\").desc_nulls_last(), F.col(\"name\").asc())\n", " .limit(3)\n", " .show()\n", ")\n", "\n", "# 3. Bottom 3\n", "print(\"=== Bottom 3 Lowest Paid ===\")\n", "employeesDF.orderBy(F.col(\"salary\").asc_nulls_last()).limit(3).show()\n", "\n", "# 4. Sample 50%\n", "print(\"=== 50% Sample ===\")\n", "employeesDF.sample(withReplacement=False, fraction=0.5, seed=42).show()\n", "\n", "# 5. Earliest five sales\n", "print(\"=== First 5 Sales by Date ===\")\n", "salesDF.orderBy(F.to_date(\"sale_date\").asc()).show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 17 — repartition, coalesce, cache" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Default partitions: 8\n", "After repartition(4): 4\n", "After coalesce(2): 2\n", "After repartition(department): 1\n", "DataFrame cached\n", "DataFrame unpersisted\n" ] } ], "source": [ "# 1. Current partitions\n", "print(f\"Default partitions: {employeesDF.rdd.getNumPartitions()}\")\n", "\n", "# 2. Repartition to 4 (full shuffle)\n", "repartitioned = employeesDF.repartition(4)\n", "print(f\"After repartition(4): {repartitioned.rdd.getNumPartitions()}\")\n", "\n", "# 3. Coalesce to 2\n", "coalesced = repartitioned.coalesce(2)\n", "print(f\"After coalesce(2): {coalesced.rdd.getNumPartitions()}\")\n", "\n", "# 4. Repartition by column\n", "by_dept = employeesDF.repartition(F.col(\"department\"))\n", "print(f\"After repartition(department): {by_dept.rdd.getNumPartitions()}\")\n", "\n", "# 5. Cache and materialize\n", "cached_df = employeesDF.cache()\n", "cached_df.count()\n", "print(\"DataFrame cached\")\n", "\n", "# 6. Unpersist\n", "cached_df.unpersist()\n", "print(\"DataFrame unpersisted\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 18 — Advanced Aggregations" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== countDistinct ===\n", "+-----------+-----------------+\n", "| category|distinct_products|\n", "+-----------+-----------------+\n", "|Electronics| 4|\n", "| Furniture| 2|\n", "+-----------+-----------------+\n", "\n", "=== approx_count_distinct ===\n", "+-----------+---------------+\n", "| category|approx_distinct|\n", "+-----------+---------------+\n", "|Electronics| 4|\n", "| Furniture| 2|\n", "+-----------+---------------+\n", "\n", "=== collect_list ===\n", "+------+---------------------+\n", "|emp_id|all_products |\n", "+------+---------------------+\n", "|1 |[Laptop, Desk, Phone]|\n", "|2 |[Phone, Monitor] |\n", "|3 |[Laptop] |\n", "|5 |[Laptop] |\n", "|4 |[Chair] |\n", "|7 |[Desk] |\n", "|6 |[Keyboard] |\n", "|9 |[Monitor] |\n", "|8 |[Chair] |\n", "+------+---------------------+\n", "\n", "=== collect_set ===\n", "+------+---------------------+\n", "|emp_id|unique_products |\n", "+------+---------------------+\n", "|1 |[Laptop, Phone, Desk]|\n", "|2 |[Phone, Monitor] |\n", "|3 |[Laptop] |\n", "|5 |[Laptop] |\n", "|4 |[Chair] |\n", "|7 |[Desk] |\n", "|6 |[Keyboard] |\n", "|9 |[Monitor] |\n", "|8 |[Chair] |\n", "+------+---------------------+\n", "\n", "=== First & Last Sale ===\n", "+------+-------------+------------+\n", "|emp_id|first_product|last_product|\n", "+------+-------------+------------+\n", "| 1| Laptop| Phone|\n", "| 2| Phone| Monitor|\n", "| 3| Laptop| Laptop|\n", "| 4| Chair| Chair|\n", "| 5| Laptop| Laptop|\n", "| 6| Keyboard| Keyboard|\n", "| 7| Desk| Desk|\n", "| 8| Chair| Chair|\n", "| 9| Monitor| Monitor|\n", "+------+-------------+------------+\n" ] } ], "source": [ "# 1. countDistinct per category\n", "print(\"=== countDistinct ===\")\n", "salesDF.groupBy(\"category\").agg(\n", " F.countDistinct(\"product\").alias(\"distinct_products\")\n", ").show()\n", "\n", "# 2. approx_count_distinct\n", "print(\"=== approx_count_distinct ===\")\n", "salesDF.groupBy(\"category\").agg(\n", " F.approx_count_distinct(\"product\").alias(\"approx_distinct\")\n", ").show()\n", "\n", "# 3. collect_list\n", "print(\"=== collect_list ===\")\n", "salesDF.groupBy(\"emp_id\").agg(\n", " F.collect_list(\"product\").alias(\"all_products\")\n", ").show(truncate=False)\n", "\n", "# 4. collect_set\n", "print(\"=== collect_set ===\")\n", "salesDF.groupBy(\"emp_id\").agg(\n", " F.collect_set(\"product\").alias(\"unique_products\")\n", ").show(truncate=False)\n", "\n", "# 5. Deterministic first and last sale by date\n", "sales_ordered = salesDF.withColumn(\"sale_date\", F.to_date(\"sale_date\"))\n", "sale_window = (\n", " Window.partitionBy(\"emp_id\")\n", " .orderBy(\"sale_date\")\n", " .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)\n", ")\n", "\n", "first_last = (\n", " sales_ordered\n", " .withColumn(\"first_product\", F.first(\"product\").over(sale_window))\n", " .withColumn(\"last_product\", F.last(\"product\").over(sale_window))\n", " .select(\"emp_id\", \"first_product\", \"last_product\")\n", " .dropDuplicates([\"emp_id\"])\n", ")\n", "print(\"=== First & Last Sale ===\")\n", "first_last.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 19 — Complex Pipeline" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n", "| name| department| salary|revenue_per_emp|salary_pct_of_budget|dept_salary_rank|perf_label|\n", "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n", "| Eve|Engineering|120000.0| 6000.0| 24.0| 1| Star|\n", "|Charlie|Engineering|110000.0| 1200.0| 22.0| 2| Good|\n", "| Hank|Engineering| 98000.0| 1500.0| 19.6| 3| Good|\n", "| Alice|Engineering| 95000.0| 4550.0| 19.0| 4| Star|\n", "| Diana| HR| 60000.0| 1000.0| 40.0| 1| Low|\n", "| Grace| HR| 58000.0| 450.0| 38.67| 2| Low|\n", "| Jack| HR| NULL| 0.0| NULL| 3| Low|\n", "| Ivy| Marketing| 75000.0| 1800.0| 25.0| 1| Good|\n", "| Bob| Marketing| 72000.0| 2000.0| 24.0| 2| Good|\n", "| Frank| Marketing| 68000.0| 800.0| 22.67| 3| Low|\n", "+-------+-----------+--------+---------------+--------------------+----------------+----------+\n" ] } ], "source": [ "dept_window = Window.partitionBy(\"department\").orderBy(\n", " F.col(\"salary\").desc_nulls_last()\n", ")\n", "\n", "# Revenue per employee\n", "revenue_df = (\n", " salesDF\n", " .withColumn(\"revenue\", F.col(\"price\") * F.col(\"quantity\"))\n", " .groupBy(\"emp_id\")\n", " .agg(F.round(F.sum(\"revenue\"), 2).alias(\"revenue_per_emp\"))\n", ")\n", "\n", "pipeline = (\n", " employeesDF\n", " .join(revenue_df, [\"emp_id\"], \"left\")\n", " .na.fill({\"revenue_per_emp\": 0.0})\n", " .join(\n", " departmentsDF,\n", " employeesDF[\"department\"] == departmentsDF[\"dept_name\"],\n", " \"left\",\n", " )\n", " .withColumn(\n", " \"salary_pct_of_budget\",\n", " F.round((F.col(\"salary\") / F.col(\"budget\")) * 100, 2),\n", " )\n", " .withColumn(\"dept_salary_rank\", F.rank().over(dept_window))\n", " .withColumn(\n", " \"perf_label\",\n", " F.when(F.col(\"revenue_per_emp\") > 3000, \"Star\")\n", " .when(F.col(\"revenue_per_emp\") > 1000, \"Good\")\n", " .otherwise(\"Low\"),\n", " )\n", " .select(\n", " \"name\", \"department\", \"salary\", \"revenue_per_emp\",\n", " \"salary_pct_of_budget\", \"dept_salary_rank\", \"perf_label\",\n", " )\n", " .orderBy(\"department\", \"dept_salary_rank\")\n", ")\n", "\n", "pipeline.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Solution 20 — Spark SQL" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Highest Avg Salary Dept ===\n", "+-----------+----------+\n", "| department|avg_salary|\n", "+-----------+----------+\n", "|Engineering| 105750.0|\n", "+-----------+----------+\n", "\n", "=== Above Dept Average ===\n", "+-------+-----------+--------+--------+\n", "| name| department| salary|dept_avg|\n", "+-------+-----------+--------+--------+\n", "| Bob| Marketing| 72000.0|71666.67|\n", "|Charlie|Engineering|110000.0|105750.0|\n", "| Eve|Engineering|120000.0|105750.0|\n", "| Diana| HR| 60000.0| 59000.0|\n", "| Ivy| Marketing| 75000.0|71666.67|\n", "+-------+-----------+--------+--------+\n", "\n", "=== Top 2 per Dept (SQL Window) ===\n", "+-------+-----------+--------+---+\n", "| name| department| salary| rn|\n", "+-------+-----------+--------+---+\n", "| Eve|Engineering|120000.0| 1|\n", "|Charlie|Engineering|110000.0| 2|\n", "| Diana| HR| 60000.0| 1|\n", "| Grace| HR| 58000.0| 2|\n", "| Ivy| Marketing| 75000.0| 1|\n", "| Bob| Marketing| 72000.0| 2|\n", "+-------+-----------+--------+---+\n", "\n", "=== Revenue per Dept ===\n", "+-----------+-------------+-------------+\n", "| department| dept_head|total_revenue|\n", "+-----------+-------------+-------------+\n", "|Engineering|Alice Johnson| 13250.0|\n", "| Marketing| Bob Williams| 4600.0|\n", "| HR| Carol Smith| 1450.0|\n", "+-----------+-------------+-------------+\n", "\n", "=== Salary > 50% of Budget (CTE) ===\n", "+-----------+------------+--------+-------------+\n", "| department|total_salary| budget|pct_of_budget|\n", "+-----------+------------+--------+-------------+\n", "|Engineering| 423000.0|500000.0| 84.6|\n", "| HR| 118000.0|150000.0| 78.7|\n", "| Marketing| 215000.0|300000.0| 71.7|\n", "+-----------+------------+--------+-------------+\n" ] } ], "source": [ "# 1. Department with the highest average salary\n", "print(\"=== Highest Avg Salary Dept ===\")\n", "spark.sql(\"\"\"\n", " SELECT department, ROUND(AVG(salary), 2) AS avg_salary\n", " FROM employees\n", " GROUP BY department\n", " ORDER BY avg_salary DESC\n", " LIMIT 1\n", "\"\"\").show()\n", "\n", "# 2. Employees earning more than their department average\n", "print(\"=== Above Dept Average ===\")\n", "spark.sql(\"\"\"\n", " SELECT e.name, e.department, e.salary, ROUND(d.avg_sal, 2) AS dept_avg\n", " FROM employees e\n", " JOIN (\n", " SELECT department, AVG(salary) AS avg_sal\n", " FROM employees\n", " GROUP BY department\n", " ) d ON e.department = d.department\n", " WHERE e.salary > d.avg_sal\n", "\"\"\").show()\n", "\n", "# 3. Top two earners per department\n", "print(\"=== Top 2 per Dept (SQL Window) ===\")\n", "spark.sql(\"\"\"\n", " SELECT name, department, salary, rn FROM (\n", " SELECT name, department, salary,\n", " ROW_NUMBER() OVER (\n", " PARTITION BY department ORDER BY salary DESC\n", " ) AS rn\n", " FROM employees\n", " ) ranked\n", " WHERE rn <= 2\n", "\"\"\").show()\n", "\n", "# 4. Total revenue per department using all three tables\n", "print(\"=== Revenue per Dept ===\")\n", "spark.sql(\"\"\"\n", " SELECT e.department, d.dept_head,\n", " ROUND(SUM(s.price * s.quantity), 2) AS total_revenue\n", " FROM employees e\n", " JOIN sales s ON e.emp_id = s.emp_id\n", " JOIN departments d ON e.department = d.dept_name\n", " GROUP BY e.department, d.dept_head\n", " ORDER BY total_revenue DESC\n", "\"\"\").show()\n", "\n", "# 5. CTE: departments where salary exceeds 50% of budget\n", "print(\"=== Salary > 50% of Budget (CTE) ===\")\n", "spark.sql(\"\"\"\n", " WITH dept_salary AS (\n", " SELECT e.department, SUM(e.salary) AS total_salary, d.budget\n", " FROM employees e\n", " JOIN departments d ON e.department = d.dept_name\n", " GROUP BY e.department, d.budget\n", " )\n", " SELECT department,\n", " ROUND(total_salary, 0) AS total_salary,\n", " budget,\n", " ROUND(total_salary / budget * 100, 1) AS pct_of_budget\n", " FROM dept_salary\n", " WHERE total_salary > 0.5 * budget\n", " ORDER BY pct_of_budget DESC\n", "\"\"\").show()" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (PySpark)", "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.11" } }, "nbformat": 4, "nbformat_minor": 4 }