{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Spark Scala 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": 1, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "\u001b[32mimport \u001b[39m\u001b[36m$ivy.$\n", "\u001b[39m\n", "\u001b[32mimport \u001b[39m\u001b[36m$ivy.$ \u001b[39m" ] }, "execution_count": 1, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import $ivy.`org.apache.spark::spark-sql:3.5.0`\n", "import $ivy.`org.apache.spark::spark-core:3.5.0`" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Setup — Run This First" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties\n", "26/03/18 15:03:05 WARN Utils: Your hostname, Prasoons-MacBook-Air.local resolves to a loopback address: 127.0.0.1; using 192.168.0.105 instead (on interface en0)\n", "26/03/18 15:03:05 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n", "26/03/18 15:03:05 INFO SparkContext: Running Spark version 3.5.0\n", "26/03/18 15:03:05 INFO SparkContext: OS info Mac OS X, 15.5, aarch64\n", "26/03/18 15:03:05 INFO SparkContext: Java version 11.0.30\n", "26/03/18 15:03:05 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "26/03/18 15:03:05 INFO ResourceUtils: ==============================================================\n", "26/03/18 15:03:05 INFO ResourceUtils: No custom resources configured for spark.driver.\n", "26/03/18 15:03:05 INFO ResourceUtils: ==============================================================\n", "26/03/18 15:03:05 INFO SparkContext: Submitted application: SparkPractice\n", "26/03/18 15:03:05 INFO ResourceProfile: Default ResourceProfile created, executor resources: Map(cores -> name: cores, amount: 1, script: , vendor: , memory -> name: memory, amount: 1024, script: , vendor: , offHeap -> name: offHeap, amount: 0, script: , vendor: ), task resources: Map(cpus -> name: cpus, amount: 1.0)\n", "26/03/18 15:03:05 INFO ResourceProfile: Limiting resource is cpu\n", "26/03/18 15:03:05 INFO ResourceProfileManager: Added ResourceProfile id: 0\n", "26/03/18 15:03:05 INFO SecurityManager: Changing view acls to: prasoonparashar\n", "26/03/18 15:03:05 INFO SecurityManager: Changing modify acls to: prasoonparashar\n", "26/03/18 15:03:05 INFO SecurityManager: Changing view acls groups to:\n", "26/03/18 15:03:05 INFO SecurityManager: Changing modify acls groups to:\n", "26/03/18 15:03:05 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: prasoonparashar; groups with view permissions: EMPTY; users with modify permissions: prasoonparashar; groups with modify permissions: EMPTY\n", "26/03/18 15:03:05 INFO Utils: Successfully started service 'sparkDriver'on port 58546.\n", "26/03/18 15:03:05 INFO SparkEnv: Registering MapOutputTracker\n", "26/03/18 15:03:05 INFO SparkEnv: Registering BlockManagerMaster\n", "26/03/18 15:03:05 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information\n", "26/03/18 15:03:05 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up\n", "26/03/18 15:03:05 INFO SparkEnv: Registering BlockManagerMasterHeartbeat\n", "26/03/18 15:03:05 INFO DiskBlockManager: Created local directory at /private/var/folders/kb/kx_wv25j447fmykqllrw6jlh0000gn/T/blockmgr-6f67b7ff-6c11-483f-8fb9-fdd1e85d0a00\n", "26/03/18 15:03:05 INFO MemoryStore: MemoryStore started with capacity 2.2 GiB\n", "26/03/18 15:03:05 INFO SparkEnv: Registering OutputCommitCoordinator\n", "26/03/18 15:03:05 INFO JettyUtils: Start Jetty 0.0.0.0:4040 for SparkUI\n", "26/03/18 15:03:05 INFO Utils: Successfully started service 'SparkUI'on port 4040.\n", "26/03/18 15:03:05 INFO Executor: Starting executor ID driver on host 192.168.0.105\n", "26/03/18 15:03:05 INFO Executor: OS info Mac OS X, 15.5, aarch64\n", "26/03/18 15:03:05 INFO Executor: Java version 11.0.30\n", "26/03/18 15:03:05 INFO Executor: Starting executor with user classpath (userClassPathFirst = false): ''\n", "26/03/18 15:03:05 INFO Executor: Created or updated repl class loader org.apache.spark.util.MutableURLClassLoader@3e54df58 for default.\n", "26/03/18 15:03:05 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService'on port 58547.\n", "26/03/18 15:03:05 INFO NettyBlockTransferService: Server created on 192.168.0.105:58547\n", "26/03/18 15:03:05 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy\n", "26/03/18 15:03:05 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.0.105, 58547, None)\n", "26/03/18 15:03:05 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.0.105:58547 with 2.2 GiB RAM, BlockManagerId(driver, 192.168.0.105, 58547, None)\n", "26/03/18 15:03:05 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.0.105, 58547, None)\n", "26/03/18 15:03:05 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.0.105, 58547, None)\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Spark Ready! Version: 3.5.0\n" ] }, { "data": { "text/plain": [ "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark.sql.SparkSession\n", "\u001b[39m\n", "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark.sql.functions._\n", "\u001b[39m\n", "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark.sql.expressions.Window\n", "\u001b[39m\n", "\u001b[32mimport \u001b[39m\u001b[36morg.apache.spark.sql.types._\n", "\n", "\u001b[39m\n", "\u001b[36mspark\u001b[39m: \u001b[32mSparkSession\u001b[39m = org.apache.spark.sql.SparkSession@38c8b1bf\n", "\u001b[32mimport \u001b[39m\u001b[36mspark.implicits._\n", "\n", "\u001b[39m" ] }, "execution_count": 2, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import org.apache.spark.sql.SparkSession\n", "import org.apache.spark.sql.functions._\n", "import org.apache.spark.sql.expressions.Window\n", "import org.apache.spark.sql.types._\n", "\n", "val spark = SparkSession.builder()\n", " .appName(\"SparkPractice\")\n", " .master(\"local[*]\")\n", " .getOrCreate()\n", "\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "import spark.implicits._\n", "\n", "println(\"Spark Ready! Version: \" + spark.version)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Dummy Data Setup" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ " All DataFrames Ready!\n", "\n", "--- Employees ---\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", "| 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| NULL| 3|2021-03-17|\n", "+------+-------+-----------+-------+--------+---------+----------+\n", "\n", "\n", "--- Sales ---\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", "| 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", "\n", "\n", "--- Departments ---\n", "+-----------+-------------+--------+\n", "| dept_name| dept_head| budget|\n", "+-----------+-------------+--------+\n", "|Engineering|Alice Johnson|500000.0|\n", "| Marketing| Bob Williams|300000.0|\n", "| HR| Carol Smith|150000.0|\n", "| Finance| David Brown|200000.0|\n", "+-----------+-------------+--------+\n", "\n", "\n", "--- Skills ---\n", "+------+--------------------+\n", "|emp_id| skills|\n", "+------+--------------------+\n", "| 1| [Scala, Spark, SQL]|\n", "| 2| [Marketing, SEO]|\n", "| 3|[Java, Scala, Kaf...|\n", "| 4| [HR, Recruiting]|\n", "| 5|[Scala, Python, M...|\n", "+------+--------------------+\n", "\n" ] }, { "data": { "text/plain": [ "\u001b[36memployeesDF\u001b[39m: \u001b[32morg\u001b[39m.\u001b[32mapache\u001b[39m.\u001b[32mspark\u001b[39m.\u001b[32msql\u001b[39m.\u001b[32mpackage\u001b[39m.\u001b[32mDataFrame\u001b[39m = [emp_id: int, name: string ... 5 more fields]\n", "\u001b[36msalesDF\u001b[39m: \u001b[32morg\u001b[39m.\u001b[32mapache\u001b[39m.\u001b[32mspark\u001b[39m.\u001b[32msql\u001b[39m.\u001b[32mpackage\u001b[39m.\u001b[32mDataFrame\u001b[39m = [sale_id: int, emp_id: int ... 5 more fields]\n", "\u001b[36mdepartmentsDF\u001b[39m: \u001b[32morg\u001b[39m.\u001b[32mapache\u001b[39m.\u001b[32mspark\u001b[39m.\u001b[32msql\u001b[39m.\u001b[32mpackage\u001b[39m.\u001b[32mDataFrame\u001b[39m = [dept_name: string, dept_head: string ... 1 more field]\n", "\u001b[36mskillsDF\u001b[39m: \u001b[32morg\u001b[39m.\u001b[32mapache\u001b[39m.\u001b[32mspark\u001b[39m.\u001b[32msql\u001b[39m.\u001b[32mpackage\u001b[39m.\u001b[32mDataFrame\u001b[39m = [emp_id: int, skills: array]" ] }, "execution_count": 3, "metadata": {}, "output_type": "execute_result" } ], "source": [ "// ─── employees DataFrame ───────────────────────────────────────────────────\n", "val employeesDF = Seq(\n", " (1, \"Alice\", \"Engineering\", \"India\", Some(95000.0), 5, \"2019-03-15\"),\n", " (2, \"Bob\", \"Marketing\", \"US\", Some(72000.0), 3, \"2021-06-01\"),\n", " (3, \"Charlie\", \"Engineering\", \"US\", Some(110000.0), 8, \"2016-01-20\"),\n", " (4, \"Diana\", \"HR\", \"India\", Some(60000.0), 2, \"2022-09-10\"),\n", " (5, \"Eve\", \"Engineering\", \"Germany\", Some(120000.0), 10,\"2014-07-04\"),\n", " (6, \"Frank\", \"Marketing\", \"India\", Some(68000.0), 4, \"2020-11-30\"),\n", " (7, \"Grace\", \"HR\", \"US\", Some(58000.0), 1, \"2023-02-14\"),\n", " (8, \"Hank\", \"Engineering\", \"Germany\", Some(98000.0), 6, \"2018-05-22\"),\n", " (9, \"Ivy\", \"Marketing\", \"Germany\", Some(75000.0), 5, \"2019-08-09\"),\n", " (10, \"Jack\", \"HR\", \"US\", None: Option[Double], 3, \"2021-03-17\")\n", ").toDF(\"emp_id\", \"name\", \"department\", \"country\", \"salary\", \"years_exp\", \"join_date\")\n", "\n", "// ─── sales DataFrame ───────────────────────────────────────────────────────\n", "val salesDF = Seq(\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", ").toDF(\"sale_id\", \"emp_id\", \"product\", \"category\", \"price\", \"quantity\", \"sale_date\")\n", "\n", "// ─── departments DataFrame ────────────────────────────────────────────────\n", "val departmentsDF = Seq(\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", ").toDF(\"dept_name\", \"dept_head\", \"budget\")\n", "\n", "// ─── skills DataFrame (array column) ─────────────────────────────────────\n", "val skillsDF = Seq(\n", " (1, Array(\"Scala\", \"Spark\", \"SQL\")),\n", " (2, Array(\"Marketing\", \"SEO\")),\n", " (3, Array(\"Java\", \"Scala\", \"Kafka\", \"Spark\")),\n", " (4, Array(\"HR\", \"Recruiting\")),\n", " (5, Array(\"Scala\", \"Python\", \"ML\", \"Spark\"))\n", ").toDF(\"emp_id\", \"skills\")\n", "\n", "employeesDF.createOrReplaceTempView(\"employees\")\n", "salesDF.createOrReplaceTempView(\"sales\")\n", "departmentsDF.createOrReplaceTempView(\"departments\")\n", "skillsDF.createOrReplaceTempView(\"skills\")\n", "\n", "println(\"All DataFrames Ready!\")\n", "println(\"\\n--- Employees ---\")\n", "employeesDF.show()\n", "println(\"\\n--- Sales ---\")\n", "salesDF.show()\n", "println(\"\\n--- Departments ---\")\n", "departmentsDF.show()\n", "println(\"\\n--- Skills ---\")\n", "skillsDF.show()" ] }, { "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", "```" ] }, { "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", "```" ] }, { "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", "| Bob| Marketing| Phone| 800.0| 1|2024-01-07|\n", "| Alice|Engineering| Desk| 450.0| 3|2024-01-10|\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", "| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|\n", "| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|\n", "| Grace| HR| Desk| 450.0| 1|2024-03-05|\n", "| Alice|Engineering| Phone| 800.0| 1|2024-03-10|\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", "```" ] }, { "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", "```" ] }, { "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", "| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 3200.0|\n", "| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 4550.0|\n", "| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 5750.0|\n", "| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 6750.0|\n", "| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 12750.0|\n", "| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 13950.0|\n", "| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 14750.0|\n", "| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 15200.0|\n", "| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 16000.0|\n", "| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 17500.0|\n", "| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 19300.0|\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", "```" ] }, { "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 na.drop (subset salary+country): 9 rows\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", "```" ] }, { "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", "```" ] }, { "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", "| name| skill|\n", "+-------+----------+\n", "| Alice| SQL|\n", "| Alice| Spark|\n", "| Alice| Scala|\n", "| Bob| SEO|\n", "| Bob| Marketing|\n", "|Charlie| Spark|\n", "|Charlie| Kafka|\n", "|Charlie| Scala|\n", "|Charlie| Java|\n", "| Diana|Recruiting|\n", "| Diana| HR|\n", "| Eve| Spark|\n", "| Eve| ML|\n", "| Eve| Python|\n", "| Eve| Scala|\n", "+-------+----------+\n", "\n", "=== Re-aggregated ===\n", "+-------+---------------------------+\n", "|name |skills_list |\n", "+-------+---------------------------+\n", "|Alice |[SQL, Spark, Scala] |\n", "|Bob |[SEO, Marketing] |\n", "|Charlie|[Spark, Kafka, Scala, Java]|\n", "|Diana |[Recruiting, HR] |\n", "|Eve |[Spark, ML, Python, Scala] |\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", "| 1| 3200.0| 1350.0|\n", "| 6| 800.0| 0.0|\n", "| 3| 1200.0| 0.0|\n", "| 5| 6000.0| 0.0|\n", "| 9| 1800.0| 0.0|\n", "| 4| 0.0| 1000.0|\n", "| 8| 0.0| 1500.0|\n", "| 7| 0.0| 450.0|\n", "| 2| 2000.0| 0.0|\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", "| Alice| 95000.0| 5|\n", "|Charlie|110000.0| 8|\n", "| Eve|120000.0| 10|\n", "| Hank| 98000.0| 6|\n", "+-------+--------+---------+\n", "\n", "=== Except: High Salary but NOT Senior ===\n", "+----+------+---------+\n", "|name|salary|years_exp|\n", "+----+------+---------+\n", "+----+------+---------+\n", "```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## Q13 — UDF (User Defined Function)\n", "\n", "1. Write a UDF `gradeEmployee` that takes `salary: Double` 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| N/A|\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", "```" ] }, { "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: 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 = false)\n", " |-- join_date: string (nullable = true)\n", "\n", "=== Casted Employees 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: integer (nullable = true)\n", " |-- years_exp: integer (nullable = false)\n", " |-- join_date: date (nullable = true)\n", "\n", "root\n", " |-- sale_id: integer (nullable = false)\n", " |-- emp_id: integer (nullable = false)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = false)\n", " |-- quantity: integer (nullable = false)\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", "```" ] }, { "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", "```" ] }, { "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(dept): 1\n", "DataFrame cached\n", "DataFrame unpersisted\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", "|6 |[Keyboard] |\n", "|7 |[Desk] |\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", "|6 |[Keyboard] |\n", "|7 |[Desk] |\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", "```" ] }, { "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", "```" ] }, { "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", "| Diana| HR| 60000.0| 59000.0|\n", "| Eve|Engineering|120000.0|105750.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", "```" ] }, { "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": [ "val q1 = employeesDF\n", " .select(\"name\", \"department\", \"salary\")\n", " .filter($\"department\" === \"Engineering\" && $\"salary\" > 90000)\n", " .withColumn(\"salary_inr\", $\"salary\" * 83)\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": [ "val q2 = employeesDF\n", " .groupBy(\"department\")\n", " .agg(\n", " count(\"emp_id\").as(\"total_employees\"),\n", " round(avg(\"salary\"), 2).as(\"avg_salary\"),\n", " max(\"salary\").as(\"max_salary\"),\n", " min(\"salary\").as(\"min_salary\"),\n", " sum(\"salary\").as(\"total_salary\")\n", " )\n", " .orderBy($\"avg_salary\".desc)\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", "| Bob| Marketing| Phone| 800.0| 1|2024-01-07|\n", "| Alice|Engineering| Desk| 450.0| 3|2024-01-10|\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", "| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|\n", "| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|\n", "| Grace| HR| Desk| 450.0| 1|2024-03-05|\n", "| Alice|Engineering| Phone| 800.0| 1|2024-03-10|\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", "val innerJoin = employeesDF\n", " .join(salesDF, \"emp_id\")\n", " .select(\"name\", \"department\", \"product\", \"price\", \"quantity\", \"sale_date\")\n", "println(\"=== Inner Join ===\")\n", "innerJoin.show()\n", "\n", "// 2. Left Join + filter null (employees with NO sales)\n", "val noSales = employeesDF\n", " .join(salesDF, Seq(\"emp_id\"), \"left\")\n", " .filter($\"sale_id\".isNull)\n", " .select(\"emp_id\", \"name\", \"department\")\n", "println(\"=== No Sales (Left + null filter) ===\")\n", "noSales.show()\n", "\n", "// 3. Anti Join\n", "val antiJoin = employeesDF\n", " .join(salesDF, Seq(\"emp_id\"), \"anti\")\n", " .select(\"emp_id\", \"name\", \"department\")\n", "println(\"=== Anti Join ===\")\n", "antiJoin.show()\n", "\n", "// 4. Departments with NO employees\n", "val deptNoEmp = departmentsDF\n", " .join(employeesDF, departmentsDF(\"dept_name\") === employeesDF(\"department\"), \"left\")\n", " .filter($\"emp_id\".isNull)\n", " .select(\"dept_name\", \"dept_head\")\n", "println(\"=== Departments with no Employees ===\")\n", "deptNoEmp.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": [ "val windowSpec = Window.partitionBy(\"department\").orderBy($\"salary\".desc)\n", "\n", "val ranked = employeesDF\n", " .withColumn(\"rank\", rank().over(windowSpec))\n", " .withColumn(\"dense_rank\", dense_rank().over(windowSpec))\n", " .withColumn(\"row_number\", row_number().over(windowSpec))\n", "\n", "println(\"=== All Ranks ===\")\n", "ranked.select(\"name\", \"department\", \"salary\", \"rank\", \"dense_rank\", \"row_number\").show()\n", "\n", "println(\"=== Top 2 per Department ===\")\n", "ranked.filter($\"dense_rank\" <= 2)\n", " .select(\"name\", \"department\", \"salary\", \"dense_rank\").show()" ] }, { "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", "| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 3200.0|\n", "| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 4550.0|\n", "| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 5750.0|\n", "| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 6750.0|\n", "| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 12750.0|\n", "| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 13950.0|\n", "| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 14750.0|\n", "| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 15200.0|\n", "| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 16000.0|\n", "| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 17500.0|\n", "| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 19300.0|\n", "+------+--------+------+--------+----------+---------------+---------------+---------------+\n" ] } ], "source": [ "val salesWithDate = salesDF.withColumn(\"sale_date\", to_date($\"sale_date\"))\n", "\n", "val empWindow = Window.partitionBy(\"emp_id\").orderBy(\"sale_date\")\n", "val globalWindow = Window.orderBy(\"sale_date\").rowsBetween(Window.unboundedPreceding, Window.currentRow)\n", "\n", "val q5 = salesWithDate\n", " .withColumn(\"prev_sale_price\", lag(\"price\", 1).over(empWindow))\n", " .withColumn(\"next_sale_price\", lead(\"price\", 1).over(empWindow))\n", " .withColumn(\"revenue\", $\"price\" * $\"quantity\")\n", " .withColumn(\"running_revenue\", sum(\"revenue\").over(globalWindow))\n", "\n", "q5.select(\"emp_id\", \"product\", \"price\", \"quantity\", \"sale_date\",\n", " \"prev_sale_price\", \"next_sale_price\", \"running_revenue\").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": [ "val q6 = employeesDF\n", " .withColumn(\"salary_band\",\n", " when($\"salary\".isNull, \"Unknown\")\n", " .when($\"salary\" >= 100000, \"High\")\n", " .when($\"salary\" >= 70000, \"Mid\")\n", " .otherwise(\"Low\")\n", " )\n", " .withColumn(\"experience_level\",\n", " when($\"years_exp\" >= 7, \"Senior\")\n", " .when($\"years_exp\" >= 4, \"Mid\")\n", " .otherwise(\"Junior\")\n", " )\n", "\n", "q6.select(\"name\", \"salary\", \"salary_band\", \"years_exp\", \"experience_level\").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 na.drop (subset salary+country): 9 rows\n" ] } ], "source": [ "// 1. Count nulls\n", "val nullCount = employeesDF.filter($\"salary\".isNull).count()\n", "println(s\"Null salary count: $nullCount\")\n", "\n", "// 2. Fill null with avg\n", "val avgSalary = employeesDF.agg(avg(\"salary\")).first().getDouble(0)\n", "val filled = employeesDF.na.fill(Map(\"salary\" -> avgSalary))\n", "println(\"=== Filled with avg ===\")\n", "filled.select(\"name\", \"salary\").show()\n", "\n", "// 3. coalesce\n", "val withCoalesce = employeesDF.withColumn(\"salary\", coalesce($\"salary\", lit(60000.0)))\n", "println(\"=== Coalesce ===\")\n", "withCoalesce.select(\"name\", \"salary\").show()\n", "\n", "// 4. dropna - any column null\n", "val dropAll = employeesDF.na.drop()\n", "println(s\"After na.drop (any): ${dropAll.count()} rows\")\n", "\n", "// 5. dropna - subset\n", "val dropSubset = employeesDF.na.drop(Array(\"salary\", \"country\"))\n", "println(s\"After na.drop (subset salary+country): ${dropSubset.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": [ "val q8 = employeesDF\n", " .withColumn(\"name_upper\", upper($\"name\"))\n", " .withColumn(\"dept_lower\", lower($\"department\"))\n", " .withColumn(\"name_dept\", concat($\"name\", lit(\" (\"), $\"department\", lit(\")\")))\n", " .withColumn(\"name_code\", substring($\"name\", 1, 3))\n", " .withColumn(\"has_a\", lower($\"name\").contains(\"a\"))\n", " .withColumn(\"country_code\", regexp_replace(regexp_replace($\"country\", \"India\", \"IN\"), \"^US$\", \"USA\"))\n", "\n", "q8.select(\"name\", \"name_upper\", \"dept_lower\", \"name_dept\", \"name_code\", \"has_a\", \"country_code\").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": [ "val q9 = employeesDF\n", " .withColumn(\"join_date\", to_date($\"join_date\", \"yyyy-MM-dd\"))\n", " .withColumn(\"join_year\", year($\"join_date\"))\n", " .withColumn(\"join_month\", month($\"join_date\"))\n", " .withColumn(\"join_day\", dayofmonth($\"join_date\"))\n", " .withColumn(\"tenure_days\", datediff(current_date(), $\"join_date\"))\n", " .withColumn(\"tenure_years\", round($\"tenure_days\" / 365.0, 1))\n", " .withColumn(\"probation_end\", date_add($\"join_date\", 30))\n", "\n", "q9.select(\"name\", \"join_date\", \"join_year\", \"join_month\", \"join_day\",\n", " \"tenure_days\", \"tenure_years\", \"probation_end\").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", "| name| skill|\n", "+-------+----------+\n", "| Alice| SQL|\n", "| Alice| Spark|\n", "| Alice| Scala|\n", "| Bob| SEO|\n", "| Bob| Marketing|\n", "|Charlie| Spark|\n", "|Charlie| Kafka|\n", "|Charlie| Scala|\n", "|Charlie| Java|\n", "| Diana|Recruiting|\n", "| Diana| HR|\n", "| Eve| Spark|\n", "| Eve| ML|\n", "| Eve| Python|\n", "| Eve| Scala|\n", "+-------+----------+\n", "\n", "=== Re-aggregated ===\n", "+-------+---------------------------+\n", "|name |skills_list |\n", "+-------+---------------------------+\n", "|Alice |[SQL, Spark, Scala] |\n", "|Bob |[SEO, Marketing] |\n", "|Charlie|[Spark, Kafka, Scala, Java]|\n", "|Diana |[Recruiting, HR] |\n", "|Eve |[Spark, ML, Python, Scala] |\n", "+-------+---------------------------+\n" ] } ], "source": [ "// 1. Explode\n", "val explodedDF = skillsDF.withColumn(\"skill\", explode($\"skills\"))\n", "println(\"=== Exploded ===\")\n", "explodedDF.select(\"emp_id\", \"skill\").show()\n", "\n", "// 2. Size\n", "val withSize = skillsDF.withColumn(\"skill_count\", size($\"skills\"))\n", "println(\"=== Skill Count ===\")\n", "withSize.show()\n", "\n", "// 3. array_contains\n", "val sparkDevs = skillsDF.filter(array_contains($\"skills\", \"Spark\"))\n", "println(\"=== Spark Developers ===\")\n", "sparkDevs.show()\n", "\n", "// 4. Join with employees\n", "val withNames = explodedDF.join(employeesDF, \"emp_id\").select(\"name\", \"skill\")\n", "println(\"=== Name + Skills ===\")\n", "withNames.show()\n", "\n", "// 5. collect_list\n", "val reaggregated = withNames.groupBy(\"name\").agg(collect_list(\"skill\").as(\"skills_list\"))\n", "println(\"=== Re-aggregated ===\")\n", "reaggregated.show(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", "| 1| 3200.0| 1350.0|\n", "| 6| 800.0| 0.0|\n", "| 3| 1200.0| 0.0|\n", "| 5| 6000.0| 0.0|\n", "| 9| 1800.0| 0.0|\n", "| 4| 0.0| 1000.0|\n", "| 8| 0.0| 1500.0|\n", "| 7| 0.0| 450.0|\n", "| 2| 2000.0| 0.0|\n", "+------+-----------+---------+\n" ] } ], "source": [ "val q11 = salesDF\n", " .withColumn(\"revenue\", $\"price\" * $\"quantity\")\n", " .groupBy(\"emp_id\")\n", " .pivot(\"category\", Seq(\"Electronics\", \"Furniture\"))\n", " .agg(round(sum(\"revenue\"), 2))\n", " .na.fill(0)\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", "| Alice| 95000.0| 5|\n", "|Charlie|110000.0| 8|\n", "| Eve|120000.0| 10|\n", "| Hank| 98000.0| 6|\n", "+-------+--------+---------+\n", "\n", "=== Except: High Salary but NOT Senior ===\n", "+----+------+---------+\n", "|name|salary|years_exp|\n", "+----+------+---------+\n", "+----+------+---------+\n" ] } ], "source": [ "val engDF = employeesDF.filter($\"department\" === \"Engineering\")\n", "val mktDF = employeesDF.filter($\"department\" === \"Marketing\")\n", "\n", "// Union\n", "println(\"=== Union ===\")\n", "engDF.union(mktDF).show()\n", "\n", "// UnionByName (reordered mktDF)\n", "val mktReordered = mktDF.select(\"department\", \"name\", \"emp_id\", \"country\", \"salary\", \"years_exp\", \"join_date\")\n", "// unionByName resolves by name not position\n", "println(\"=== UnionByName ===\")\n", "engDF.unionByName(mktReordered).show()\n", "\n", "// Intersect & Except\n", "val highSalaryDF = employeesDF.filter($\"salary\" > 80000)\n", "val seniorDF = employeesDF.filter($\"years_exp\" >= 5)\n", "\n", "println(\"=== Intersect: High Salary AND Senior ===\")\n", "highSalaryDF.intersect(seniorDF).select(\"name\", \"salary\", \"years_exp\").show()\n", "\n", "println(\"=== Except: High Salary but NOT Senior ===\")\n", "highSalaryDF.except(seniorDF).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| N/A|\n", "+-------+-----------+--------+-----+\n" ] } ], "source": [ "// Define UDFs\n", "val gradeEmployee = udf((salary: java.lang.Double) => {\n", " if (salary == null) \"N/A\"\n", " else if (salary >= 100000) \"A\"\n", " else if (salary >= 75000) \"B\"\n", " else if (salary >= 50000) \"C\"\n", " else \"D\"\n", "})\n", "\n", "val maskName = udf((name: String) => {\n", " if (name == null || name.isEmpty) \"\"\n", " else name.head.toString + \"***\"\n", "})\n", "\n", "// Register for SQL use\n", "spark.udf.register(\"gradeEmployee\", gradeEmployee)\n", "spark.udf.register(\"maskName\", maskName)\n", "\n", "val q13 = employeesDF\n", " .withColumn(\"grade\", gradeEmployee($\"salary\"))\n", " .withColumn(\"masked_name\", maskName($\"name\"))\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", "println(\"=== Distinct Products ===\")\n", "salesDF.select(\"product\").distinct().show()\n", "\n", "// 2. Create duplicates\n", "val dupeDF = salesDF.union(salesDF)\n", "println(s\"With duplicates: ${dupeDF.count()} rows\")\n", "\n", "// 3. dropDuplicates full row\n", "val deduped = dupeDF.dropDuplicates()\n", "println(s\"After dropDuplicates(): ${deduped.count()} rows\")\n", "\n", "// 4. dropDuplicates on subset\n", "val onePerEmpProduct = dupeDF.dropDuplicates(Seq(\"emp_id\", \"product\"))\n", "println(s\"After dropDuplicates(emp_id, product): ${onePerEmpProduct.count()} rows\")\n", "onePerEmpProduct.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: 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 = false)\n", " |-- join_date: string (nullable = true)\n", "\n", "=== Casted Employees 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: integer (nullable = true)\n", " |-- years_exp: integer (nullable = false)\n", " |-- join_date: date (nullable = true)\n", "\n", "root\n", " |-- sale_id: integer (nullable = false)\n", " |-- emp_id: integer (nullable = false)\n", " |-- product: string (nullable = true)\n", " |-- category: string (nullable = true)\n", " |-- price: double (nullable = false)\n", " |-- quantity: integer (nullable = false)\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 schema\n", "println(\"=== Employees Schema ===\")\n", "employeesDF.printSchema()\n", "\n", "// 2. Cast join_date to DateType\n", "val empCasted = employeesDF\n", " .withColumn(\"join_date\", $\"join_date\".cast(DateType))\n", " .withColumn(\"salary\", $\"salary\".cast(IntegerType))\n", "\n", "println(\"=== Casted Employees Schema ===\")\n", "empCasted.printSchema()\n", "\n", "// 3. Cast sale_date to Timestamp\n", "val salesCasted = salesDF.withColumn(\"sale_date\", $\"sale_date\".cast(TimestampType))\n", "salesCasted.printSchema()\n", "\n", "// 4. Manual StructType schema\n", "val manualSchema = StructType(Seq(\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", "println(\"=== Manual Schema ===\")\n", "println(manualSchema.treeString)" ] }, { "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 & 2. Sort + top 3\n", "println(\"=== Top 3 Highest Paid ===\")\n", "employeesDF.orderBy($\"salary\".desc_nulls_last, $\"name\".asc).limit(3).show()\n", "\n", "// 3. Bottom 3\n", "println(\"=== Bottom 3 Lowest Paid ===\")\n", "employeesDF.orderBy($\"salary\".asc_nulls_last).limit(3).show()\n", "\n", "// 4. Sample 50%\n", "println(\"=== 50% Sample ===\")\n", "employeesDF.sample(withReplacement = false, fraction = 0.5, seed = 42).show()\n", "\n", "// 5. salesDF ordered by date\n", "println(\"=== First 5 Sales by Date ===\")\n", "salesDF.orderBy(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(dept): 1\n", "DataFrame cached\n", "DataFrame unpersisted\n" ] } ], "source": [ "// 1. Current partitions\n", "println(s\"Default partitions: ${employeesDF.rdd.getNumPartitions}\")\n", "\n", "// 2. Repartition to 4 (full shuffle)\n", "val repartitioned = employeesDF.repartition(4)\n", "println(s\"After repartition(4): ${repartitioned.rdd.getNumPartitions}\")\n", "\n", "// 3. Coalesce to 2 (no full shuffle)\n", "val coalesced = repartitioned.coalesce(2)\n", "println(s\"After coalesce(2): ${coalesced.rdd.getNumPartitions}\")\n", "\n", "// 4. Repartition by column\n", "val byDept = employeesDF.repartition($\"department\")\n", "println(s\"After repartition(dept): ${byDept.rdd.getNumPartitions}\")\n", "\n", "// 5. Cache\n", "val cachedDF = employeesDF.cache()\n", "cachedDF.count() // trigger materialization\n", "println(\"DataFrame cached\")\n", "\n", "// 6. Unpersist\n", "cachedDF.unpersist()\n", "println(\"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", "|6 |[Keyboard] |\n", "|7 |[Desk] |\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", "|6 |[Keyboard] |\n", "|7 |[Desk] |\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", "println(\"=== countDistinct ===\")\n", "salesDF.groupBy(\"category\")\n", " .agg(countDistinct(\"product\").as(\"distinct_products\"))\n", " .show()\n", "\n", "// 2. approx_count_distinct\n", "println(\"=== approx_count_distinct ===\")\n", "salesDF.groupBy(\"category\")\n", " .agg(approx_count_distinct(\"product\").as(\"approx_distinct\"))\n", " .show()\n", "\n", "// 3. collect_list\n", "println(\"=== collect_list ===\")\n", "salesDF.groupBy(\"emp_id\")\n", " .agg(collect_list(\"product\").as(\"all_products\"))\n", " .show(false)\n", "\n", "// 4. collect_set\n", "println(\"=== collect_set ===\")\n", "salesDF.groupBy(\"emp_id\")\n", " .agg(collect_set(\"product\").as(\"unique_products\"))\n", " .show(false)\n", "\n", "// 5. first / last by sale_date\n", "val salesOrdered = salesDF.withColumn(\"sale_date\", to_date($\"sale_date\"))\n", "val empSalesWindow = Window.partitionBy(\"emp_id\").orderBy(\"sale_date\")\n", "\n", "println(\"=== First & Last Sale ===\")\n", "salesOrdered.groupBy(\"emp_id\")\n", " .agg(\n", " first(\"product\").as(\"first_product\"),\n", " last(\"product\").as(\"last_product\")\n", " ).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": [ "val deptWindow = Window.partitionBy(\"department\").orderBy($\"salary\".desc_nulls_last)\n", "\n", "// Revenue per employee\n", "val revenueDF = salesDF\n", " .withColumn(\"revenue\", $\"price\" * $\"quantity\")\n", " .groupBy(\"emp_id\")\n", " .agg(round(sum(\"revenue\"), 2).as(\"revenue_per_emp\"))\n", "\n", "val pipeline = employeesDF\n", " .join(revenueDF, Seq(\"emp_id\"), \"left\")\n", " .na.fill(Map(\"revenue_per_emp\" -> 0.0))\n", " .join(departmentsDF, employeesDF(\"department\") === departmentsDF(\"dept_name\"), \"left\")\n", " .withColumn(\"salary_pct_of_budget\",\n", " round(($\"salary\" / $\"budget\") * 100, 2))\n", " .withColumn(\"dept_salary_rank\",\n", " rank().over(deptWindow))\n", " .withColumn(\"perf_label\",\n", " when($\"revenue_per_emp\" > 3000, \"Star\")\n", " .when($\"revenue_per_emp\" > 1000, \"Good\")\n", " .otherwise(\"Low\")\n", " )\n", " .select(\"name\", \"department\", \"salary\", \"revenue_per_emp\",\n", " \"salary_pct_of_budget\", \"dept_salary_rank\", \"perf_label\")\n", " .orderBy(\"department\", \"dept_salary_rank\")\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", "| Diana| HR| 60000.0| 59000.0|\n", "| Eve|Engineering|120000.0|105750.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. Dept with highest avg salary\n", "println(\"=== 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 dept avg\n", "println(\"=== 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 2 per dept using ROW_NUMBER\n", "println(\"=== 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 (PARTITION BY department ORDER BY salary DESC) AS rn\n", " FROM employees\n", " ) ranked\n", " WHERE rn <= 2\n", "\"\"\").show()\n", "\n", "// 4. Total revenue per department (3-table join)\n", "println(\"=== 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 — depts where total salary > 50% of budget\n", "println(\"=== 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": "Scala", "language": "scala", "name": "scala" }, "language_info": { "codemirror_mode": "text/x-scala", "file_extension": ".sc", "mimetype": "text/x-scala", "name": "scala", "nbconvert_exporter": "script", "version": "2.13.10" } }, "nbformat": 4, "nbformat_minor": 4 }