Preparing chapter
Loading your data design lesson
The curriculum shell is ready while the requested chapter is being prepared. You can wait here or return to the design library.
Preparing chapter
The curriculum shell is ready while the requested chapter is being prepared. You can wait here or return to the design library.
Scala + Spark Practice
spark-scala-practice-problems.ipynb
110 cells · 47 code · 26 saved outputs
Trusted · exact notebook preview
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.
Dummy data is pre-loaded. Write your transformations in the // YOUR CODE HERE cells. Solutions are provided at the bottom of this notebook.import $ivy.`org.apache.spark::spark-sql:3.5.0`
import $ivy.`org.apache.spark::spark-core:3.5.0`import $ivy.$
import $ivy.$ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.types._
val spark = SparkSession.builder()
.appName("SparkPractice")
.master("local[*]")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
println("Spark Ready! Version: " + spark.version)Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties
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)
26/03/18 15:03:05 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
26/03/18 15:03:05 INFO SparkContext: Running Spark version 3.5.0
26/03/18 15:03:05 INFO SparkContext: OS info Mac OS X, 15.5, aarch64
26/03/18 15:03:05 INFO SparkContext: Java version 11.0.30
26/03/18 15:03:05 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
26/03/18 15:03:05 INFO ResourceUtils: ==============================================================
26/03/18 15:03:05 INFO ResourceUtils: No custom resources configured for spark.driver.
26/03/18 15:03:05 INFO ResourceUtils: ==============================================================
26/03/18 15:03:05 INFO SparkContext: Submitted application: SparkPractice
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)
26/03/18 15:03:05 INFO ResourceProfile: Limiting resource is cpu
26/03/18 15:03:05 INFO ResourceProfileManager: Added ResourceProfile id: 0
26/03/18 15:03:05 INFO SecurityManager: Changing view acls to: prasoonparashar
26/03/18 15:03:05 INFO SecurityManager: Changing modify acls to: prasoonparashar
26/03/18 15:03:05 INFO SecurityManager: Changing view acls groups to:
26/03/18 15:03:05 INFO SecurityManager: Changing modify acls groups to:
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
26/03/18 15:03:05 INFO Utils: Successfully started service 'sparkDriver'on port 58546.
26/03/18 15:03:05 INFO SparkEnv: Registering MapOutputTracker
26/03/18 15:03:05 INFO SparkEnv: Registering BlockManagerMaster
26/03/18 15:03:05 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
26/03/18 15:03:05 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
26/03/18 15:03:05 INFO SparkEnv: Registering BlockManagerMasterHeartbeat
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
26/03/18 15:03:05 INFO MemoryStore: MemoryStore started with capacity 2.2 GiB
26/03/18 15:03:05 INFO SparkEnv: Registering OutputCommitCoordinator
26/03/18 15:03:05 INFO JettyUtils: Start Jetty 0.0.0.0:4040 for SparkUI
26/03/18 15:03:05 INFO Utils: Successfully started service 'SparkUI'on port 4040.
26/03/18 15:03:05 INFO Executor: Starting executor ID driver on host 192.168.0.105
26/03/18 15:03:05 INFO Executor: OS info Mac OS X, 15.5, aarch64
26/03/18 15:03:05 INFO Executor: Java version 11.0.30
26/03/18 15:03:05 INFO Executor: Starting executor with user classpath (userClassPathFirst = false): ''
26/03/18 15:03:05 INFO Executor: Created or updated repl class loader org.apache.spark.util.MutableURLClassLoader@3e54df58 for default.
26/03/18 15:03:05 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService'on port 58547.
26/03/18 15:03:05 INFO NettyBlockTransferService: Server created on 192.168.0.105:58547
26/03/18 15:03:05 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
26/03/18 15:03:05 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.0.105, 58547, None)
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)
26/03/18 15:03:05 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.0.105, 58547, None)
26/03/18 15:03:05 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.0.105, 58547, None)
Spark Ready! Version: 3.5.0
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.types._
spark: SparkSession = org.apache.spark.sql.SparkSession@38c8b1bf
import spark.implicits._
// ─── employees DataFrame ───────────────────────────────────────────────────
val employeesDF = Seq(
(1, "Alice", "Engineering", "India", Some(95000.0), 5, "2019-03-15"),
(2, "Bob", "Marketing", "US", Some(72000.0), 3, "2021-06-01"),
(3, "Charlie", "Engineering", "US", Some(110000.0), 8, "2016-01-20"),
(4, "Diana", "HR", "India", Some(60000.0), 2, "2022-09-10"),
(5, "Eve", "Engineering", "Germany", Some(120000.0), 10,"2014-07-04"),
(6, "Frank", "Marketing", "India", Some(68000.0), 4, "2020-11-30"),
(7, "Grace", "HR", "US", Some(58000.0), 1, "2023-02-14"),
(8, "Hank", "Engineering", "Germany", Some(98000.0), 6, "2018-05-22"),
(9, "Ivy", "Marketing", "Germany", Some(75000.0), 5, "2019-08-09"),
(10, "Jack", "HR", "US", None: Option[Double], 3, "2021-03-17")
).toDF("emp_id", "name", "department", "country", "salary", "years_exp", "join_date")
// ─── sales DataFrame ───────────────────────────────────────────────────────
val salesDF = Seq(
(101, 1, "Laptop", "Electronics", 1200.0, 2, "2024-01-05"),
(102, 2, "Phone", "Electronics", 800.0, 1, "2024-01-07"),
(103, 1, "Desk", "Furniture", 450.0, 3, "2024-01-10"),
(104, 3, "Laptop", "Electronics", 1200.0, 1, "2024-01-12"),
(105, 4, "Chair", "Furniture", 250.0, 4, "2024-02-01"),
(106, 5, "Laptop", "Electronics", 1200.0, 5, "2024-02-03"),
(107, 2, "Monitor", "Electronics", 600.0, 2, "2024-02-15"),
(108, 6, "Keyboard", "Electronics", 80.0,10, "2024-03-01"),
(109, 7, "Desk", "Furniture", 450.0, 1, "2024-03-05"),
(110, 1, "Phone", "Electronics", 800.0, 1, "2024-03-10"),
(111, 8, "Chair", "Furniture", 250.0, 6, "2024-03-15"),
(112, 9, "Monitor", "Electronics", 600.0, 3, "2024-04-01")
).toDF("sale_id", "emp_id", "product", "category", "price", "quantity", "sale_date")
// ─── departments DataFrame ────────────────────────────────────────────────
val departmentsDF = Seq(
("Engineering", "Alice Johnson", 500000.0),
("Marketing", "Bob Williams", 300000.0),
("HR", "Carol Smith", 150000.0),
("Finance", "David Brown", 200000.0) // No employees in this dept
).toDF("dept_name", "dept_head", "budget")
// ─── skills DataFrame (array column) ─────────────────────────────────────
val skillsDF = Seq(
(1, Array("Scala", "Spark", "SQL")),
(2, Array("Marketing", "SEO")),
(3, Array("Java", "Scala", "Kafka", "Spark")),
(4, Array("HR", "Recruiting")),
(5, Array("Scala", "Python", "ML", "Spark"))
).toDF("emp_id", "skills")
employeesDF.createOrReplaceTempView("employees")
salesDF.createOrReplaceTempView("sales")
departmentsDF.createOrReplaceTempView("departments")
skillsDF.createOrReplaceTempView("skills")
println("All DataFrames Ready!")
println("\n--- Employees ---")
employeesDF.show()
println("\n--- Sales ---")
salesDF.show()
println("\n--- Departments ---")
departmentsDF.show()
println("\n--- Skills ---")
skillsDF.show() All DataFrames Ready!
--- Employees ---
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|
| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 4| Diana| HR| India| 60000.0| 2|2022-09-10|
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 7| Grace| HR| US| 58000.0| 1|2023-02-14|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
| 10| Jack| HR| US| NULL| 3|2021-03-17|
+------+-------+-----------+-------+--------+---------+----------+
--- Sales ---
+-------+------+--------+-----------+------+--------+----------+
|sale_id|emp_id| product| category| price|quantity| sale_date|
+-------+------+--------+-----------+------+--------+----------+
| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|
| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|
| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|
| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|
| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|
| 106| 5| Laptop|Electronics|1200.0| 5|2024-02-03|
| 107| 2| Monitor|Electronics| 600.0| 2|2024-02-15|
| 108| 6|Keyboard|Electronics| 80.0| 10|2024-03-01|
| 109| 7| Desk| Furniture| 450.0| 1|2024-03-05|
| 110| 1| Phone|Electronics| 800.0| 1|2024-03-10|
| 111| 8| Chair| Furniture| 250.0| 6|2024-03-15|
| 112| 9| Monitor|Electronics| 600.0| 3|2024-04-01|
+-------+------+--------+-----------+------+--------+----------+
--- Departments ---
+-----------+-------------+--------+
| dept_name| dept_head| budget|
+-----------+-------------+--------+
|Engineering|Alice Johnson|500000.0|
| Marketing| Bob Williams|300000.0|
| HR| Carol Smith|150000.0|
| Finance| David Brown|200000.0|
+-----------+-------------+--------+
--- Skills ---
+------+--------------------+
|emp_id| skills|
+------+--------------------+
| 1| [Scala, Spark, SQL]|
| 2| [Marketing, SEO]|
| 3|[Java, Scala, Kaf...|
| 4| [HR, Recruiting]|
| 5|[Scala, Python, M...|
+------+--------------------+
employeesDF: org.apache.spark.sql.package.DataFrame = [emp_id: int, name: string ... 5 more fields]
salesDF: org.apache.spark.sql.package.DataFrame = [sale_id: int, emp_id: int ... 5 more fields]
departmentsDF: org.apache.spark.sql.package.DataFrame = [dept_name: string, dept_head: string ... 1 more field]
skillsDF: org.apache.spark.sql.package.DataFrame = [emp_id: int, skills: array<string>]From employeesDF:
name, department, salaryEngineering with salary > 90000salary_inr = salary × 83 (USD to INR)// YOUR CODE HERE+-------+-----------+--------+----------+
| name| department| salary|salary_inr|
+-------+-----------+--------+----------+
| Alice|Engineering| 95000.0| 7885000.0|
|Charlie|Engineering|110000.0| 9130000.0|
| Eve|Engineering|120000.0| 9960000.0|
| Hank|Engineering| 98000.0| 8134000.0|
+-------+-----------+--------+----------+From employeesDF, for each department compute:
count)avg)sum)Round avg to 2 decimal places. Sort by avg salary descending.
// YOUR CODE HERE+-----------+---------------+----------+----------+----------+------------+
| department|total_employees|avg_salary|max_salary|min_salary|total_salary|
+-----------+---------------+----------+----------+----------+------------+
|Engineering| 4| 105750.0| 120000.0| 95000.0| 423000.0|
| Marketing| 3| 71666.67| 75000.0| 68000.0| 215000.0|
| HR| 3| 59000.0| 60000.0| 58000.0| 118000.0|
+-----------+---------------+----------+----------+----------+------------+employeesDF with salesDF on emp_id → show employee name and their salesemployeesDF with salesDF → find employees who made NO sales (anti-join pattern using left + filter null)anti join type directlydepartmentsDF with employeesDF → show departments with no employees// YOUR CODE HERE=== Inner Join ===
+-------+-----------+--------+------+--------+----------+
| name| department| product| price|quantity| sale_date|
+-------+-----------+--------+------+--------+----------+
| Alice|Engineering| Laptop|1200.0| 2|2024-01-05|
| Bob| Marketing| Phone| 800.0| 1|2024-01-07|
| Alice|Engineering| Desk| 450.0| 3|2024-01-10|
|Charlie|Engineering| Laptop|1200.0| 1|2024-01-12|
| Diana| HR| Chair| 250.0| 4|2024-02-01|
| Eve|Engineering| Laptop|1200.0| 5|2024-02-03|
| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|
| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|
| Grace| HR| Desk| 450.0| 1|2024-03-05|
| Alice|Engineering| Phone| 800.0| 1|2024-03-10|
| Hank|Engineering| Chair| 250.0| 6|2024-03-15|
| Ivy| Marketing| Monitor| 600.0| 3|2024-04-01|
+-------+-----------+--------+------+--------+----------+
=== No Sales (Left + null filter) ===
+------+----+----------+
|emp_id|name|department|
+------+----+----------+
| 10|Jack| HR|
+------+----+----------+
=== Anti Join ===
+------+----+----------+
|emp_id|name|department|
+------+----+----------+
| 10|Jack| HR|
+------+----+----------+
=== Departments with no Employees ===
+---------+-----------+
|dept_name| dept_head|
+---------+-----------+
| Finance|David Brown|
+---------+-----------+From employeesDF:
department (highest first)rank, dense_rank, row_number — all partitioned by department, ordered by salary desc// YOUR CODE HERE=== All Ranks ===
+-------+-----------+--------+----+----------+----------+
| name| department| salary|rank|dense_rank|row_number|
+-------+-----------+--------+----+----------+----------+
| Eve|Engineering|120000.0| 1| 1| 1|
|Charlie|Engineering|110000.0| 2| 2| 2|
| Hank|Engineering| 98000.0| 3| 3| 3|
| Alice|Engineering| 95000.0| 4| 4| 4|
| Diana| HR| 60000.0| 1| 1| 1|
| Grace| HR| 58000.0| 2| 2| 2|
| Jack| HR| NULL| 3| 3| 3|
| Ivy| Marketing| 75000.0| 1| 1| 1|
| Bob| Marketing| 72000.0| 2| 2| 2|
| Frank| Marketing| 68000.0| 3| 3| 3|
+-------+-----------+--------+----+----------+----------+
=== Top 2 per Department ===
+-------+-----------+--------+----------+
| name| department| salary|dense_rank|
+-------+-----------+--------+----------+
| Eve|Engineering|120000.0| 1|
|Charlie|Engineering|110000.0| 2|
| Diana| HR| 60000.0| 1|
| Grace| HR| 58000.0| 2|
| Ivy| Marketing| 75000.0| 1|
| Bob| Marketing| 72000.0| 2|
+-------+-----------+--------+----------+From salesDF (cast sale_date as Date, order by it per emp_id):
prev_sale_price = lag of price by 1, partitioned by emp_idnext_sale_price = lead of price by 1, partitioned by emp_idrunning_revenue = cumulative sum of price * quantity ordered by sale_date// YOUR CODE HERE+------+--------+------+--------+----------+---------------+---------------+---------------+
|emp_id| product| price|quantity| sale_date|prev_sale_price|next_sale_price|running_revenue|
+------+--------+------+--------+----------+---------------+---------------+---------------+
| 1| Laptop|1200.0| 2|2024-01-05| NULL| 450.0| 2400.0|
| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 3200.0|
| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 4550.0|
| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 5750.0|
| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 6750.0|
| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 12750.0|
| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 13950.0|
| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 14750.0|
| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 15200.0|
| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 16000.0|
| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 17500.0|
| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 19300.0|
+------+--------+------+--------+----------+---------------+---------------+---------------+From employeesDF, create a new column salary_band:
"High" if salary >= 100000"Mid" if salary >= 70000"Low" if salary >= 0"Unknown" if salary is nullAlso create experience_level:
"Senior" if years_exp >= 7"Mid" if years_exp >= 4"Junior" otherwise// YOUR CODE HERE+-------+--------+-----------+---------+----------------+
| name| salary|salary_band|years_exp|experience_level|
+-------+--------+-----------+---------+----------------+
| Alice| 95000.0| Mid| 5| Mid|
| Bob| 72000.0| Mid| 3| Junior|
|Charlie|110000.0| High| 8| Senior|
| Diana| 60000.0| Low| 2| Junior|
| Eve|120000.0| High| 10| Senior|
| Frank| 68000.0| Low| 4| Mid|
| Grace| 58000.0| Low| 1| Junior|
| Hank| 98000.0| Mid| 6| Mid|
| Ivy| 75000.0| Mid| 5| Mid|
| Jack| NULL| Unknown| 3| Junior|
+-------+--------+-----------+---------+----------------+From employeesDF:
salary is nullcoalesce to replace null salary with 60000.0 as fallback// YOUR CODE HERENull salary count: 1
=== Filled with avg ===
+-------+--------+
| name| salary|
+-------+--------+
| Alice| 95000.0|
| Bob| 72000.0|
|Charlie|110000.0|
| Diana| 60000.0|
| Eve|120000.0|
| Frank| 68000.0|
| Grace| 58000.0|
| Hank| 98000.0|
| Ivy| 75000.0|
| Jack| 84000.0|
+-------+--------+
=== Coalesce ===
+-------+--------+
| name| salary|
+-------+--------+
| Alice| 95000.0|
| Bob| 72000.0|
|Charlie|110000.0|
| Diana| 60000.0|
| Eve|120000.0|
| Frank| 68000.0|
| Grace| 58000.0|
| Hank| 98000.0|
| Ivy| 75000.0|
| Jack| 60000.0|
+-------+--------+
After na.drop (any): 9 rows
After na.drop (subset salary+country): 9 rowsFrom employeesDF:
name, lowercase departmentname and department as "Alice (Engineering)"name as name_codename contains the letter "a" (case-insensitive) → boolean column"India" with "IN" and "US" with "USA" in the country column using regexp_replace// YOUR CODE HERE// YOUR CODE HERE// YOUR CODE HERE+-------+----------+-----------+--------------------+---------+-----+------------+
| name|name_upper| dept_lower| name_dept|name_code|has_a|country_code|
+-------+----------+-----------+--------------------+---------+-----+------------+
| Alice| ALICE|engineering| Alice (Engineering)| Ali| true| IN|
| Bob| BOB| marketing| Bob (Marketing)| Bob|false| USA|
|Charlie| CHARLIE|engineering|Charlie (Engineer...| Cha| true| USA|
| Diana| DIANA| hr| Diana (HR)| Dia| true| IN|
| Eve| EVE|engineering| Eve (Engineering)| Eve|false| Germany|
| Frank| FRANK| marketing| Frank (Marketing)| Fra| true| IN|
| Grace| GRACE| hr| Grace (HR)| Gra| true| USA|
| Hank| HANK|engineering| Hank (Engineering)| Han| true| Germany|
| Ivy| IVY| marketing| Ivy (Marketing)| Ivy|false| Germany|
| Jack| JACK| hr| Jack (HR)| Jac| true| USA|
+-------+----------+-----------+--------------------+---------+-----+------------+From employeesDF (treat join_date as a date):
join_date to DateTypeyear, month, day as separate columnstenure_days = today's date minus join_datetenure_years (rounded to 1 decimal)join_date as probation_end// YOUR CODE HERE// YOUR CODE HERE+-------+----------+---------+----------+--------+-----------+------------+-------------+
| name| join_date|join_year|join_month|join_day|tenure_days|tenure_years|probation_end|
+-------+----------+---------+----------+--------+-----------+------------+-------------+
| Alice|2019-03-15| 2019| 3| 15| 2764| 7.6| 2019-04-14|
| Bob|2021-06-01| 2021| 6| 1| 1955| 5.4| 2021-07-01|
|Charlie|2016-01-20| 2016| 1| 20| 3914| 10.7| 2016-02-19|
| Diana|2022-09-10| 2022| 9| 10| 1489| 4.1| 2022-10-10|
| Eve|2014-07-04| 2014| 7| 4| 4479| 12.3| 2014-08-03|
| Frank|2020-11-30| 2020| 11| 30| 2138| 5.9| 2020-12-30|
| Grace|2023-02-14| 2023| 2| 14| 1332| 3.6| 2023-03-16|
| Hank|2018-05-22| 2018| 5| 22| 3061| 8.4| 2018-06-21|
| Ivy|2019-08-09| 2019| 8| 9| 2617| 7.2| 2019-09-08|
| Jack|2021-03-17| 2021| 3| 17| 2031| 5.6| 2021-04-16|
+-------+----------+---------+----------+--------+-----------+------------+-------------+From skillsDF:
explode the skills array → one row per skill per employeesize of array)"Spark" in their skills (array_contains)employeesDF to show employee name alongside their individual skillscollect_list — after explode and join, re-aggregate skills back per employee// YOUR CODE HERE// YOUR CODE HERE=== Exploded ===
+------+----------+
|emp_id| skill|
+------+----------+
| 1| Scala|
| 1| Spark|
| 1| SQL|
| 2| Marketing|
| 2| SEO|
| 3| Java|
| 3| Scala|
| 3| Kafka|
| 3| Spark|
| 4| HR|
| 4|Recruiting|
| 5| Scala|
| 5| Python|
| 5| ML|
| 5| Spark|
+------+----------+
=== Skill Count ===
+------+--------------------+-----------+
|emp_id| skills|skill_count|
+------+--------------------+-----------+
| 1| [Scala, Spark, SQL]| 3|
| 2| [Marketing, SEO]| 2|
| 3|[Java, Scala, Kaf...| 4|
| 4| [HR, Recruiting]| 2|
| 5|[Scala, Python, M...| 4|
+------+--------------------+-----------+
=== Spark Developers ===
+------+--------------------+
|emp_id| skills|
+------+--------------------+
| 1| [Scala, Spark, SQL]|
| 3|[Java, Scala, Kaf...|
| 5|[Scala, Python, M...|
+------+--------------------+
=== Name + Skills ===
+-------+----------+
| name| skill|
+-------+----------+
| Alice| SQL|
| Alice| Spark|
| Alice| Scala|
| Bob| SEO|
| Bob| Marketing|
|Charlie| Spark|
|Charlie| Kafka|
|Charlie| Scala|
|Charlie| Java|
| Diana|Recruiting|
| Diana| HR|
| Eve| Spark|
| Eve| ML|
| Eve| Python|
| Eve| Scala|
+-------+----------+
=== Re-aggregated ===
+-------+---------------------------+
|name |skills_list |
+-------+---------------------------+
|Alice |[SQL, Spark, Scala] |
|Bob |[SEO, Marketing] |
|Charlie|[Spark, Kafka, Scala, Java]|
|Diana |[Recruiting, HR] |
|Eve |[Spark, ML, Python, Scala] |
+-------+---------------------------+From salesDF:
category (columns: Electronics, Furniture)emp_idsum of price * quantity)// YOUR CODE HERE+------+-----------+---------+
|emp_id|Electronics|Furniture|
+------+-----------+---------+
| 1| 3200.0| 1350.0|
| 6| 800.0| 0.0|
| 3| 1200.0| 0.0|
| 5| 6000.0| 0.0|
| 9| 1800.0| 0.0|
| 4| 0.0| 1000.0|
| 8| 0.0| 1500.0|
| 7| 0.0| 450.0|
| 2| 2000.0| 0.0|
+------+-----------+---------+employeesDF: engDF (Engineering) and mktDF (Marketing)union them and show resultunionByName — demonstrate with a reordered mktDFhighSalaryDF (salary > 80000) and seniorDF (years_exp >= 5)intersect → employees both high-salary AND seniorexcept → high-salary employees who are NOT senior// YOUR CODE HERE=== Union ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|
| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
+------+-------+-----------+-------+--------+---------+----------+
=== UnionByName ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|
| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
+------+-------+-----------+-------+--------+---------+----------+
=== Intersect: High Salary AND Senior ===
+-------+--------+---------+
| name| salary|years_exp|
+-------+--------+---------+
| Alice| 95000.0| 5|
|Charlie|110000.0| 8|
| Eve|120000.0| 10|
| Hank| 98000.0| 6|
+-------+--------+---------+
=== Except: High Salary but NOT Senior ===
+----+------+---------+
|name|salary|years_exp|
+----+------+---------+
+----+------+---------+gradeEmployee that takes salary: Double and returns:"A" if >= 100000, "B" if >= 75000, "C" if >= 50000, "D" otherwiseemployeesDF as a new column grademaskName that returns first char + "***" (e.g., "Alice" → "A***")maskName to the name column// YOUR CODE HERE+-------+-----------+--------+-----+
| name|masked_name| salary|grade|
+-------+-----------+--------+-----+
| Alice| A***| 95000.0| B|
| Bob| B***| 72000.0| C|
|Charlie| C***|110000.0| A|
| Diana| D***| 60000.0| C|
| Eve| E***|120000.0| A|
| Frank| F***| 68000.0| C|
| Grace| G***| 58000.0| C|
| Hank| H***| 98000.0| B|
| Ivy| I***| 75000.0| B|
| Jack| J***| NULL| N/A|
+-------+-----------+--------+-----+salesDF, find all unique product names using distinctsalesDF with itselfdropDuplicates() to remove full row duplicatesdropDuplicates(Seq("emp_id", "product")) to keep only one sale per employee-product combo// YOUR CODE HERE=== Distinct Products ===
+--------+
| product|
+--------+
| Laptop|
| Desk|
| Phone|
| Chair|
| Monitor|
|Keyboard|
+--------+
With duplicates: 24 rows
After dropDuplicates(): 12 rows
After dropDuplicates(emp_id, product): 12 rows
+-------+------+--------+-----------+------+--------+----------+
|sale_id|emp_id| product| category| price|quantity| sale_date|
+-------+------+--------+-----------+------+--------+----------+
| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|
| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|
| 110| 1| Phone|Electronics| 800.0| 1|2024-03-10|
| 107| 2| Monitor|Electronics| 600.0| 2|2024-02-15|
| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|
| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|
| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|
| 106| 5| Laptop|Electronics|1200.0| 5|2024-02-03|
| 108| 6|Keyboard|Electronics| 80.0| 10|2024-03-01|
| 109| 7| Desk| Furniture| 450.0| 1|2024-03-05|
| 111| 8| Chair| Furniture| 250.0| 6|2024-03-15|
| 112| 9| Monitor|Electronics| 600.0| 3|2024-04-01|
+-------+------+--------+-----------+------+--------+----------+employeesDF and salesDFjoin_date (String) to DateType in employeesDFsalary to IntegerTypesale_date (String) to TimestampType in salesDFStructType + StructField and re-read the employees data with it// YOUR CODE HERE=== Employees Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: double (nullable = true)
|-- years_exp: integer (nullable = false)
|-- join_date: string (nullable = true)
=== Casted Employees Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: integer (nullable = true)
|-- years_exp: integer (nullable = false)
|-- join_date: date (nullable = true)
root
|-- sale_id: integer (nullable = false)
|-- emp_id: integer (nullable = false)
|-- product: string (nullable = true)
|-- category: string (nullable = true)
|-- price: double (nullable = false)
|-- quantity: integer (nullable = false)
|-- sale_date: timestamp (nullable = true)
=== Manual Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: double (nullable = true)
|-- years_exp: integer (nullable = true)
|-- join_date: string (nullable = true)employeesDF by salary descending, then by name ascendingsalesDF by sale_date ascending and show the first 5 records// YOUR CODE HERE=== Top 3 Highest Paid ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
+------+-------+-----------+-------+--------+---------+----------+
=== Bottom 3 Lowest Paid ===
+------+-----+----------+-------+-------+---------+----------+
|emp_id| name|department|country| salary|years_exp| join_date|
+------+-----+----------+-------+-------+---------+----------+
| 7|Grace| HR| US|58000.0| 1|2023-02-14|
| 4|Diana| HR| India|60000.0| 2|2022-09-10|
| 6|Frank| Marketing| India|68000.0| 4|2020-11-30|
+------+-----+----------+-------+-------+---------+----------+
=== 50% Sample ===
+------+-----+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-----+-----------+-------+--------+---------+----------+
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 6|Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
| 10| Jack| HR| US| NULL| 3|2021-03-17|
+------+-----+-----------+-------+--------+---------+----------+
=== First 5 Sales by Date ===
+-------+------+-------+-----------+------+--------+----------+
|sale_id|emp_id|product| category| price|quantity| sale_date|
+-------+------+-------+-----------+------+--------+----------+
| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|
| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|
| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|
| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|
| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|
+-------+------+-------+-----------+------+--------+----------+
only showing top 5 rowsemployeesDFrepartition(4) — repartition to 4 partitions (full shuffle)coalesce(2) — reduce to 2 partitions without full shufflerepartition(col("department")) — partition by department columncache() the employeesDF and trigger materialization with an action// YOUR CODE HEREDefault partitions: 8
After repartition(4): 4
After coalesce(2): 2
After repartition(dept): 1
DataFrame cached
DataFrame unpersistedFrom salesDF:
categoryapprox_count_distinct for approximate distinct countcollect_list of all product names per emp_idcollect_set of unique products per emp_id (no duplicates)first and last sale per emp_id (by sale_date)// YOUR CODE HERE=== countDistinct ===
+-----------+-----------------+
| category|distinct_products|
+-----------+-----------------+
|Electronics| 4|
| Furniture| 2|
+-----------+-----------------+
=== approx_count_distinct ===
+-----------+---------------+
| category|approx_distinct|
+-----------+---------------+
|Electronics| 4|
| Furniture| 2|
+-----------+---------------+
=== collect_list ===
+------+---------------------+
|emp_id|all_products |
+------+---------------------+
|1 |[Laptop, Desk, Phone]|
|2 |[Phone, Monitor] |
|3 |[Laptop] |
|5 |[Laptop] |
|4 |[Chair] |
|6 |[Keyboard] |
|7 |[Desk] |
|9 |[Monitor] |
|8 |[Chair] |
+------+---------------------+
=== collect_set ===
+------+---------------------+
|emp_id|unique_products |
+------+---------------------+
|1 |[Laptop, Phone, Desk]|
|2 |[Phone, Monitor] |
|3 |[Laptop] |
|5 |[Laptop] |
|4 |[Chair] |
|6 |[Keyboard] |
|7 |[Desk] |
|9 |[Monitor] |
|8 |[Chair] |
+------+---------------------+
=== First & Last Sale ===
+------+-------------+------------+
|emp_id|first_product|last_product|
+------+-------------+------------+
| 1| Laptop| Phone|
| 2| Phone| Monitor|
| 3| Laptop| Laptop|
| 4| Chair| Chair|
| 5| Laptop| Laptop|
| 6| Keyboard| Keyboard|
| 7| Desk| Desk|
| 8| Chair| Chair|
| 9| Monitor| Monitor|
+------+-------------+------------+Build a complete employee performance report combining multiple transformations:
employeesDF with salesDF on emp_id (left join to keep all employees)revenue_per_emp = sum(price × quantity) per employeedepartmentsDF on department = dept_namesalary_pct_of_budget = (salary / budget) × 100dept_salary_rank = rank within department by salaryperf_label using when/otherwise: "Star" if revenue > 3000, "Good" if > 1000, "Low" otherwise// YOUR CODE HERE+-------+-----------+--------+---------------+--------------------+----------------+----------+
| name| department| salary|revenue_per_emp|salary_pct_of_budget|dept_salary_rank|perf_label|
+-------+-----------+--------+---------------+--------------------+----------------+----------+
| Eve|Engineering|120000.0| 6000.0| 24.0| 1| Star|
|Charlie|Engineering|110000.0| 1200.0| 22.0| 2| Good|
| Hank|Engineering| 98000.0| 1500.0| 19.6| 3| Good|
| Alice|Engineering| 95000.0| 4550.0| 19.0| 4| Star|
| Diana| HR| 60000.0| 1000.0| 40.0| 1| Low|
| Grace| HR| 58000.0| 450.0| 38.67| 2| Low|
| Jack| HR| NULL| 0.0| NULL| 3| Low|
| Ivy| Marketing| 75000.0| 1800.0| 25.0| 1| Good|
| Bob| Marketing| 72000.0| 2000.0| 24.0| 2| Good|
| Frank| Marketing| 68000.0| 800.0| 22.67| 3| Low|
+-------+-----------+--------+---------------+--------------------+----------------+----------+Use Spark SQL (not DataFrame API) to:
ROW_NUMBER() OVER (PARTITION BY ...)// YOUR CODE HERE=== Highest Avg Salary Dept ===
+-----------+----------+
| department|avg_salary|
+-----------+----------+
|Engineering| 105750.0|
+-----------+----------+
=== Above Dept Average ===
+-------+-----------+--------+--------+
| name| department| salary|dept_avg|
+-------+-----------+--------+--------+
| Bob| Marketing| 72000.0|71666.67|
|Charlie|Engineering|110000.0|105750.0|
| Diana| HR| 60000.0| 59000.0|
| Eve|Engineering|120000.0|105750.0|
| Ivy| Marketing| 75000.0|71666.67|
+-------+-----------+--------+--------+
=== Top 2 per Dept (SQL Window) ===
+-------+-----------+--------+---+
| name| department| salary| rn|
+-------+-----------+--------+---+
| Eve|Engineering|120000.0| 1|
|Charlie|Engineering|110000.0| 2|
| Diana| HR| 60000.0| 1|
| Grace| HR| 58000.0| 2|
| Ivy| Marketing| 75000.0| 1|
| Bob| Marketing| 72000.0| 2|
+-------+-----------+--------+---+
=== Revenue per Dept ===
+-----------+-------------+-------------+
| department| dept_head|total_revenue|
+-----------+-------------+-------------+
|Engineering|Alice Johnson| 13250.0|
| Marketing| Bob Williams| 4600.0|
| HR| Carol Smith| 1450.0|
+-----------+-------------+-------------+
=== Salary > 50% of Budget (CTE) ===
+-----------+------------+--------+-------------+
| department|total_salary| budget|pct_of_budget|
+-----------+------------+--------+-------------+
|Engineering| 423000.0|500000.0| 84.6|
| HR| 118000.0|150000.0| 78.7|
| Marketing| 215000.0|300000.0| 71.7|
+-----------+------------+--------+-------------+val q1 = employeesDF
.select("name", "department", "salary")
.filter($"department" === "Engineering" && $"salary" > 90000)
.withColumn("salary_inr", $"salary" * 83)
q1.show()+-------+-----------+--------+----------+
| name| department| salary|salary_inr|
+-------+-----------+--------+----------+
| Alice|Engineering| 95000.0| 7885000.0|
|Charlie|Engineering|110000.0| 9130000.0|
| Eve|Engineering|120000.0| 9960000.0|
| Hank|Engineering| 98000.0| 8134000.0|
+-------+-----------+--------+----------+
val q2 = employeesDF
.groupBy("department")
.agg(
count("emp_id").as("total_employees"),
round(avg("salary"), 2).as("avg_salary"),
max("salary").as("max_salary"),
min("salary").as("min_salary"),
sum("salary").as("total_salary")
)
.orderBy($"avg_salary".desc)
q2.show()+-----------+---------------+----------+----------+----------+------------+
| department|total_employees|avg_salary|max_salary|min_salary|total_salary|
+-----------+---------------+----------+----------+----------+------------+
|Engineering| 4| 105750.0| 120000.0| 95000.0| 423000.0|
| Marketing| 3| 71666.67| 75000.0| 68000.0| 215000.0|
| HR| 3| 59000.0| 60000.0| 58000.0| 118000.0|
+-----------+---------------+----------+----------+----------+------------+
// 1. Inner Join
val innerJoin = employeesDF
.join(salesDF, "emp_id")
.select("name", "department", "product", "price", "quantity", "sale_date")
println("=== Inner Join ===")
innerJoin.show()
// 2. Left Join + filter null (employees with NO sales)
val noSales = employeesDF
.join(salesDF, Seq("emp_id"), "left")
.filter($"sale_id".isNull)
.select("emp_id", "name", "department")
println("=== No Sales (Left + null filter) ===")
noSales.show()
// 3. Anti Join
val antiJoin = employeesDF
.join(salesDF, Seq("emp_id"), "anti")
.select("emp_id", "name", "department")
println("=== Anti Join ===")
antiJoin.show()
// 4. Departments with NO employees
val deptNoEmp = departmentsDF
.join(employeesDF, departmentsDF("dept_name") === employeesDF("department"), "left")
.filter($"emp_id".isNull)
.select("dept_name", "dept_head")
println("=== Departments with no Employees ===")
deptNoEmp.show()=== Inner Join ===
+-------+-----------+--------+------+--------+----------+
| name| department| product| price|quantity| sale_date|
+-------+-----------+--------+------+--------+----------+
| Alice|Engineering| Laptop|1200.0| 2|2024-01-05|
| Bob| Marketing| Phone| 800.0| 1|2024-01-07|
| Alice|Engineering| Desk| 450.0| 3|2024-01-10|
|Charlie|Engineering| Laptop|1200.0| 1|2024-01-12|
| Diana| HR| Chair| 250.0| 4|2024-02-01|
| Eve|Engineering| Laptop|1200.0| 5|2024-02-03|
| Bob| Marketing| Monitor| 600.0| 2|2024-02-15|
| Frank| Marketing|Keyboard| 80.0| 10|2024-03-01|
| Grace| HR| Desk| 450.0| 1|2024-03-05|
| Alice|Engineering| Phone| 800.0| 1|2024-03-10|
| Hank|Engineering| Chair| 250.0| 6|2024-03-15|
| Ivy| Marketing| Monitor| 600.0| 3|2024-04-01|
+-------+-----------+--------+------+--------+----------+
=== No Sales (Left + null filter) ===
+------+----+----------+
|emp_id|name|department|
+------+----+----------+
| 10|Jack| HR|
+------+----+----------+
=== Anti Join ===
+------+----+----------+
|emp_id|name|department|
+------+----+----------+
| 10|Jack| HR|
+------+----+----------+
=== Departments with no Employees ===
+---------+-----------+
|dept_name| dept_head|
+---------+-----------+
| Finance|David Brown|
+---------+-----------+
val windowSpec = Window.partitionBy("department").orderBy($"salary".desc)
val ranked = employeesDF
.withColumn("rank", rank().over(windowSpec))
.withColumn("dense_rank", dense_rank().over(windowSpec))
.withColumn("row_number", row_number().over(windowSpec))
println("=== All Ranks ===")
ranked.select("name", "department", "salary", "rank", "dense_rank", "row_number").show()
println("=== Top 2 per Department ===")
ranked.filter($"dense_rank" <= 2)
.select("name", "department", "salary", "dense_rank").show()=== All Ranks ===
+-------+-----------+--------+----+----------+----------+
| name| department| salary|rank|dense_rank|row_number|
+-------+-----------+--------+----+----------+----------+
| Eve|Engineering|120000.0| 1| 1| 1|
|Charlie|Engineering|110000.0| 2| 2| 2|
| Hank|Engineering| 98000.0| 3| 3| 3|
| Alice|Engineering| 95000.0| 4| 4| 4|
| Diana| HR| 60000.0| 1| 1| 1|
| Grace| HR| 58000.0| 2| 2| 2|
| Jack| HR| NULL| 3| 3| 3|
| Ivy| Marketing| 75000.0| 1| 1| 1|
| Bob| Marketing| 72000.0| 2| 2| 2|
| Frank| Marketing| 68000.0| 3| 3| 3|
+-------+-----------+--------+----+----------+----------+
=== Top 2 per Department ===
+-------+-----------+--------+----------+
| name| department| salary|dense_rank|
+-------+-----------+--------+----------+
| Eve|Engineering|120000.0| 1|
|Charlie|Engineering|110000.0| 2|
| Diana| HR| 60000.0| 1|
| Grace| HR| 58000.0| 2|
| Ivy| Marketing| 75000.0| 1|
| Bob| Marketing| 72000.0| 2|
+-------+-----------+--------+----------+
val salesWithDate = salesDF.withColumn("sale_date", to_date($"sale_date"))
val empWindow = Window.partitionBy("emp_id").orderBy("sale_date")
val globalWindow = Window.orderBy("sale_date").rowsBetween(Window.unboundedPreceding, Window.currentRow)
val q5 = salesWithDate
.withColumn("prev_sale_price", lag("price", 1).over(empWindow))
.withColumn("next_sale_price", lead("price", 1).over(empWindow))
.withColumn("revenue", $"price" * $"quantity")
.withColumn("running_revenue", sum("revenue").over(globalWindow))
q5.select("emp_id", "product", "price", "quantity", "sale_date",
"prev_sale_price", "next_sale_price", "running_revenue").show()+------+--------+------+--------+----------+---------------+---------------+---------------+
|emp_id| product| price|quantity| sale_date|prev_sale_price|next_sale_price|running_revenue|
+------+--------+------+--------+----------+---------------+---------------+---------------+
| 1| Laptop|1200.0| 2|2024-01-05| NULL| 450.0| 2400.0|
| 2| Phone| 800.0| 1|2024-01-07| NULL| 600.0| 3200.0|
| 1| Desk| 450.0| 3|2024-01-10| 1200.0| 800.0| 4550.0|
| 3| Laptop|1200.0| 1|2024-01-12| NULL| NULL| 5750.0|
| 4| Chair| 250.0| 4|2024-02-01| NULL| NULL| 6750.0|
| 5| Laptop|1200.0| 5|2024-02-03| NULL| NULL| 12750.0|
| 2| Monitor| 600.0| 2|2024-02-15| 800.0| NULL| 13950.0|
| 6|Keyboard| 80.0| 10|2024-03-01| NULL| NULL| 14750.0|
| 7| Desk| 450.0| 1|2024-03-05| NULL| NULL| 15200.0|
| 1| Phone| 800.0| 1|2024-03-10| 450.0| NULL| 16000.0|
| 8| Chair| 250.0| 6|2024-03-15| NULL| NULL| 17500.0|
| 9| Monitor| 600.0| 3|2024-04-01| NULL| NULL| 19300.0|
+------+--------+------+--------+----------+---------------+---------------+---------------+
val q6 = employeesDF
.withColumn("salary_band",
when($"salary".isNull, "Unknown")
.when($"salary" >= 100000, "High")
.when($"salary" >= 70000, "Mid")
.otherwise("Low")
)
.withColumn("experience_level",
when($"years_exp" >= 7, "Senior")
.when($"years_exp" >= 4, "Mid")
.otherwise("Junior")
)
q6.select("name", "salary", "salary_band", "years_exp", "experience_level").show()+-------+--------+-----------+---------+----------------+
| name| salary|salary_band|years_exp|experience_level|
+-------+--------+-----------+---------+----------------+
| Alice| 95000.0| Mid| 5| Mid|
| Bob| 72000.0| Mid| 3| Junior|
|Charlie|110000.0| High| 8| Senior|
| Diana| 60000.0| Low| 2| Junior|
| Eve|120000.0| High| 10| Senior|
| Frank| 68000.0| Low| 4| Mid|
| Grace| 58000.0| Low| 1| Junior|
| Hank| 98000.0| Mid| 6| Mid|
| Ivy| 75000.0| Mid| 5| Mid|
| Jack| NULL| Unknown| 3| Junior|
+-------+--------+-----------+---------+----------------+
// 1. Count nulls
val nullCount = employeesDF.filter($"salary".isNull).count()
println(s"Null salary count: $nullCount")
// 2. Fill null with avg
val avgSalary = employeesDF.agg(avg("salary")).first().getDouble(0)
val filled = employeesDF.na.fill(Map("salary" -> avgSalary))
println("=== Filled with avg ===")
filled.select("name", "salary").show()
// 3. coalesce
val withCoalesce = employeesDF.withColumn("salary", coalesce($"salary", lit(60000.0)))
println("=== Coalesce ===")
withCoalesce.select("name", "salary").show()
// 4. dropna - any column null
val dropAll = employeesDF.na.drop()
println(s"After na.drop (any): ${dropAll.count()} rows")
// 5. dropna - subset
val dropSubset = employeesDF.na.drop(Array("salary", "country"))
println(s"After na.drop (subset salary+country): ${dropSubset.count()} rows")Null salary count: 1
=== Filled with avg ===
+-------+--------+
| name| salary|
+-------+--------+
| Alice| 95000.0|
| Bob| 72000.0|
|Charlie|110000.0|
| Diana| 60000.0|
| Eve|120000.0|
| Frank| 68000.0|
| Grace| 58000.0|
| Hank| 98000.0|
| Ivy| 75000.0|
| Jack| 84000.0|
+-------+--------+
=== Coalesce ===
+-------+--------+
| name| salary|
+-------+--------+
| Alice| 95000.0|
| Bob| 72000.0|
|Charlie|110000.0|
| Diana| 60000.0|
| Eve|120000.0|
| Frank| 68000.0|
| Grace| 58000.0|
| Hank| 98000.0|
| Ivy| 75000.0|
| Jack| 60000.0|
+-------+--------+
After na.drop (any): 9 rows
After na.drop (subset salary+country): 9 rows
val q8 = employeesDF
.withColumn("name_upper", upper($"name"))
.withColumn("dept_lower", lower($"department"))
.withColumn("name_dept", concat($"name", lit(" ("), $"department", lit(")")))
.withColumn("name_code", substring($"name", 1, 3))
.withColumn("has_a", lower($"name").contains("a"))
.withColumn("country_code", regexp_replace(regexp_replace($"country", "India", "IN"), "^US$", "USA"))
q8.select("name", "name_upper", "dept_lower", "name_dept", "name_code", "has_a", "country_code").show()+-------+----------+-----------+--------------------+---------+-----+------------+
| name|name_upper| dept_lower| name_dept|name_code|has_a|country_code|
+-------+----------+-----------+--------------------+---------+-----+------------+
| Alice| ALICE|engineering| Alice (Engineering)| Ali| true| IN|
| Bob| BOB| marketing| Bob (Marketing)| Bob|false| USA|
|Charlie| CHARLIE|engineering|Charlie (Engineer...| Cha| true| USA|
| Diana| DIANA| hr| Diana (HR)| Dia| true| IN|
| Eve| EVE|engineering| Eve (Engineering)| Eve|false| Germany|
| Frank| FRANK| marketing| Frank (Marketing)| Fra| true| IN|
| Grace| GRACE| hr| Grace (HR)| Gra| true| USA|
| Hank| HANK|engineering| Hank (Engineering)| Han| true| Germany|
| Ivy| IVY| marketing| Ivy (Marketing)| Ivy|false| Germany|
| Jack| JACK| hr| Jack (HR)| Jac| true| USA|
+-------+----------+-----------+--------------------+---------+-----+------------+
val q9 = employeesDF
.withColumn("join_date", to_date($"join_date", "yyyy-MM-dd"))
.withColumn("join_year", year($"join_date"))
.withColumn("join_month", month($"join_date"))
.withColumn("join_day", dayofmonth($"join_date"))
.withColumn("tenure_days", datediff(current_date(), $"join_date"))
.withColumn("tenure_years", round($"tenure_days" / 365.0, 1))
.withColumn("probation_end", date_add($"join_date", 30))
q9.select("name", "join_date", "join_year", "join_month", "join_day",
"tenure_days", "tenure_years", "probation_end").show()+-------+----------+---------+----------+--------+-----------+------------+-------------+
| name| join_date|join_year|join_month|join_day|tenure_days|tenure_years|probation_end|
+-------+----------+---------+----------+--------+-----------+------------+-------------+
| Alice|2019-03-15| 2019| 3| 15| 2764| 7.6| 2019-04-14|
| Bob|2021-06-01| 2021| 6| 1| 1955| 5.4| 2021-07-01|
|Charlie|2016-01-20| 2016| 1| 20| 3914| 10.7| 2016-02-19|
| Diana|2022-09-10| 2022| 9| 10| 1489| 4.1| 2022-10-10|
| Eve|2014-07-04| 2014| 7| 4| 4479| 12.3| 2014-08-03|
| Frank|2020-11-30| 2020| 11| 30| 2138| 5.9| 2020-12-30|
| Grace|2023-02-14| 2023| 2| 14| 1332| 3.6| 2023-03-16|
| Hank|2018-05-22| 2018| 5| 22| 3061| 8.4| 2018-06-21|
| Ivy|2019-08-09| 2019| 8| 9| 2617| 7.2| 2019-09-08|
| Jack|2021-03-17| 2021| 3| 17| 2031| 5.6| 2021-04-16|
+-------+----------+---------+----------+--------+-----------+------------+-------------+
// 1. Explode
val explodedDF = skillsDF.withColumn("skill", explode($"skills"))
println("=== Exploded ===")
explodedDF.select("emp_id", "skill").show()
// 2. Size
val withSize = skillsDF.withColumn("skill_count", size($"skills"))
println("=== Skill Count ===")
withSize.show()
// 3. array_contains
val sparkDevs = skillsDF.filter(array_contains($"skills", "Spark"))
println("=== Spark Developers ===")
sparkDevs.show()
// 4. Join with employees
val withNames = explodedDF.join(employeesDF, "emp_id").select("name", "skill")
println("=== Name + Skills ===")
withNames.show()
// 5. collect_list
val reaggregated = withNames.groupBy("name").agg(collect_list("skill").as("skills_list"))
println("=== Re-aggregated ===")
reaggregated.show(false)=== Exploded ===
+------+----------+
|emp_id| skill|
+------+----------+
| 1| Scala|
| 1| Spark|
| 1| SQL|
| 2| Marketing|
| 2| SEO|
| 3| Java|
| 3| Scala|
| 3| Kafka|
| 3| Spark|
| 4| HR|
| 4|Recruiting|
| 5| Scala|
| 5| Python|
| 5| ML|
| 5| Spark|
+------+----------+
=== Skill Count ===
+------+--------------------+-----------+
|emp_id| skills|skill_count|
+------+--------------------+-----------+
| 1| [Scala, Spark, SQL]| 3|
| 2| [Marketing, SEO]| 2|
| 3|[Java, Scala, Kaf...| 4|
| 4| [HR, Recruiting]| 2|
| 5|[Scala, Python, M...| 4|
+------+--------------------+-----------+
=== Spark Developers ===
+------+--------------------+
|emp_id| skills|
+------+--------------------+
| 1| [Scala, Spark, SQL]|
| 3|[Java, Scala, Kaf...|
| 5|[Scala, Python, M...|
+------+--------------------+
=== Name + Skills ===
+-------+----------+
| name| skill|
+-------+----------+
| Alice| SQL|
| Alice| Spark|
| Alice| Scala|
| Bob| SEO|
| Bob| Marketing|
|Charlie| Spark|
|Charlie| Kafka|
|Charlie| Scala|
|Charlie| Java|
| Diana|Recruiting|
| Diana| HR|
| Eve| Spark|
| Eve| ML|
| Eve| Python|
| Eve| Scala|
+-------+----------+
=== Re-aggregated ===
+-------+---------------------------+
|name |skills_list |
+-------+---------------------------+
|Alice |[SQL, Spark, Scala] |
|Bob |[SEO, Marketing] |
|Charlie|[Spark, Kafka, Scala, Java]|
|Diana |[Recruiting, HR] |
|Eve |[Spark, ML, Python, Scala] |
+-------+---------------------------+
val q11 = salesDF
.withColumn("revenue", $"price" * $"quantity")
.groupBy("emp_id")
.pivot("category", Seq("Electronics", "Furniture"))
.agg(round(sum("revenue"), 2))
.na.fill(0)
q11.show()+------+-----------+---------+
|emp_id|Electronics|Furniture|
+------+-----------+---------+
| 1| 3200.0| 1350.0|
| 6| 800.0| 0.0|
| 3| 1200.0| 0.0|
| 5| 6000.0| 0.0|
| 9| 1800.0| 0.0|
| 4| 0.0| 1000.0|
| 8| 0.0| 1500.0|
| 7| 0.0| 450.0|
| 2| 2000.0| 0.0|
+------+-----------+---------+
val engDF = employeesDF.filter($"department" === "Engineering")
val mktDF = employeesDF.filter($"department" === "Marketing")
// Union
println("=== Union ===")
engDF.union(mktDF).show()
// UnionByName (reordered mktDF)
val mktReordered = mktDF.select("department", "name", "emp_id", "country", "salary", "years_exp", "join_date")
// unionByName resolves by name not position
println("=== UnionByName ===")
engDF.unionByName(mktReordered).show()
// Intersect & Except
val highSalaryDF = employeesDF.filter($"salary" > 80000)
val seniorDF = employeesDF.filter($"years_exp" >= 5)
println("=== Intersect: High Salary AND Senior ===")
highSalaryDF.intersect(seniorDF).select("name", "salary", "years_exp").show()
println("=== Except: High Salary but NOT Senior ===")
highSalaryDF.except(seniorDF).select("name", "salary", "years_exp").show()=== Union ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|
| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
+------+-------+-----------+-------+--------+---------+----------+
=== UnionByName ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 1| Alice|Engineering| India| 95000.0| 5|2019-03-15|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 2| Bob| Marketing| US| 72000.0| 3|2021-06-01|
| 6| Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
+------+-------+-----------+-------+--------+---------+----------+
=== Intersect: High Salary AND Senior ===
+-------+--------+---------+
| name| salary|years_exp|
+-------+--------+---------+
| Alice| 95000.0| 5|
|Charlie|110000.0| 8|
| Eve|120000.0| 10|
| Hank| 98000.0| 6|
+-------+--------+---------+
=== Except: High Salary but NOT Senior ===
+----+------+---------+
|name|salary|years_exp|
+----+------+---------+
+----+------+---------+
// Define UDFs
val gradeEmployee = udf((salary: java.lang.Double) => {
if (salary == null) "N/A"
else if (salary >= 100000) "A"
else if (salary >= 75000) "B"
else if (salary >= 50000) "C"
else "D"
})
val maskName = udf((name: String) => {
if (name == null || name.isEmpty) ""
else name.head.toString + "***"
})
// Register for SQL use
spark.udf.register("gradeEmployee", gradeEmployee)
spark.udf.register("maskName", maskName)
val q13 = employeesDF
.withColumn("grade", gradeEmployee($"salary"))
.withColumn("masked_name", maskName($"name"))
q13.select("name", "masked_name", "salary", "grade").show()+-------+-----------+--------+-----+
| name|masked_name| salary|grade|
+-------+-----------+--------+-----+
| Alice| A***| 95000.0| B|
| Bob| B***| 72000.0| C|
|Charlie| C***|110000.0| A|
| Diana| D***| 60000.0| C|
| Eve| E***|120000.0| A|
| Frank| F***| 68000.0| C|
| Grace| G***| 58000.0| C|
| Hank| H***| 98000.0| B|
| Ivy| I***| 75000.0| B|
| Jack| J***| NULL| N/A|
+-------+-----------+--------+-----+
// 1. Distinct products
println("=== Distinct Products ===")
salesDF.select("product").distinct().show()
// 2. Create duplicates
val dupeDF = salesDF.union(salesDF)
println(s"With duplicates: ${dupeDF.count()} rows")
// 3. dropDuplicates full row
val deduped = dupeDF.dropDuplicates()
println(s"After dropDuplicates(): ${deduped.count()} rows")
// 4. dropDuplicates on subset
val onePerEmpProduct = dupeDF.dropDuplicates(Seq("emp_id", "product"))
println(s"After dropDuplicates(emp_id, product): ${onePerEmpProduct.count()} rows")
onePerEmpProduct.show()=== Distinct Products ===
+--------+
| product|
+--------+
| Laptop|
| Desk|
| Phone|
| Chair|
| Monitor|
|Keyboard|
+--------+
With duplicates: 24 rows
After dropDuplicates(): 12 rows
After dropDuplicates(emp_id, product): 12 rows
+-------+------+--------+-----------+------+--------+----------+
|sale_id|emp_id| product| category| price|quantity| sale_date|
+-------+------+--------+-----------+------+--------+----------+
| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|
| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|
| 110| 1| Phone|Electronics| 800.0| 1|2024-03-10|
| 107| 2| Monitor|Electronics| 600.0| 2|2024-02-15|
| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|
| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|
| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|
| 106| 5| Laptop|Electronics|1200.0| 5|2024-02-03|
| 108| 6|Keyboard|Electronics| 80.0| 10|2024-03-01|
| 109| 7| Desk| Furniture| 450.0| 1|2024-03-05|
| 111| 8| Chair| Furniture| 250.0| 6|2024-03-15|
| 112| 9| Monitor|Electronics| 600.0| 3|2024-04-01|
+-------+------+--------+-----------+------+--------+----------+
// 1. Print schema
println("=== Employees Schema ===")
employeesDF.printSchema()
// 2. Cast join_date to DateType
val empCasted = employeesDF
.withColumn("join_date", $"join_date".cast(DateType))
.withColumn("salary", $"salary".cast(IntegerType))
println("=== Casted Employees Schema ===")
empCasted.printSchema()
// 3. Cast sale_date to Timestamp
val salesCasted = salesDF.withColumn("sale_date", $"sale_date".cast(TimestampType))
salesCasted.printSchema()
// 4. Manual StructType schema
val manualSchema = StructType(Seq(
StructField("emp_id", IntegerType, nullable = false),
StructField("name", StringType, nullable = true),
StructField("department",StringType, nullable = true),
StructField("country", StringType, nullable = true),
StructField("salary", DoubleType, nullable = true),
StructField("years_exp", IntegerType, nullable = true),
StructField("join_date", StringType, nullable = true)
))
println("=== Manual Schema ===")
println(manualSchema.treeString)=== Employees Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: double (nullable = true)
|-- years_exp: integer (nullable = false)
|-- join_date: string (nullable = true)
=== Casted Employees Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: integer (nullable = true)
|-- years_exp: integer (nullable = false)
|-- join_date: date (nullable = true)
root
|-- sale_id: integer (nullable = false)
|-- emp_id: integer (nullable = false)
|-- product: string (nullable = true)
|-- category: string (nullable = true)
|-- price: double (nullable = false)
|-- quantity: integer (nullable = false)
|-- sale_date: timestamp (nullable = true)
=== Manual Schema ===
root
|-- emp_id: integer (nullable = false)
|-- name: string (nullable = true)
|-- department: string (nullable = true)
|-- country: string (nullable = true)
|-- salary: double (nullable = true)
|-- years_exp: integer (nullable = true)
|-- join_date: string (nullable = true)
// 1 & 2. Sort + top 3
println("=== Top 3 Highest Paid ===")
employeesDF.orderBy($"salary".desc_nulls_last, $"name".asc).limit(3).show()
// 3. Bottom 3
println("=== Bottom 3 Lowest Paid ===")
employeesDF.orderBy($"salary".asc_nulls_last).limit(3).show()
// 4. Sample 50%
println("=== 50% Sample ===")
employeesDF.sample(withReplacement = false, fraction = 0.5, seed = 42).show()
// 5. salesDF ordered by date
println("=== First 5 Sales by Date ===")
salesDF.orderBy(to_date($"sale_date").asc).show(5)=== Top 3 Highest Paid ===
+------+-------+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-------+-----------+-------+--------+---------+----------+
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 3|Charlie|Engineering| US|110000.0| 8|2016-01-20|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
+------+-------+-----------+-------+--------+---------+----------+
=== Bottom 3 Lowest Paid ===
+------+-----+----------+-------+-------+---------+----------+
|emp_id| name|department|country| salary|years_exp| join_date|
+------+-----+----------+-------+-------+---------+----------+
| 7|Grace| HR| US|58000.0| 1|2023-02-14|
| 4|Diana| HR| India|60000.0| 2|2022-09-10|
| 6|Frank| Marketing| India|68000.0| 4|2020-11-30|
+------+-----+----------+-------+-------+---------+----------+
=== 50% Sample ===
+------+-----+-----------+-------+--------+---------+----------+
|emp_id| name| department|country| salary|years_exp| join_date|
+------+-----+-----------+-------+--------+---------+----------+
| 5| Eve|Engineering|Germany|120000.0| 10|2014-07-04|
| 6|Frank| Marketing| India| 68000.0| 4|2020-11-30|
| 8| Hank|Engineering|Germany| 98000.0| 6|2018-05-22|
| 9| Ivy| Marketing|Germany| 75000.0| 5|2019-08-09|
| 10| Jack| HR| US| NULL| 3|2021-03-17|
+------+-----+-----------+-------+--------+---------+----------+
=== First 5 Sales by Date ===
+-------+------+-------+-----------+------+--------+----------+
|sale_id|emp_id|product| category| price|quantity| sale_date|
+-------+------+-------+-----------+------+--------+----------+
| 101| 1| Laptop|Electronics|1200.0| 2|2024-01-05|
| 102| 2| Phone|Electronics| 800.0| 1|2024-01-07|
| 103| 1| Desk| Furniture| 450.0| 3|2024-01-10|
| 104| 3| Laptop|Electronics|1200.0| 1|2024-01-12|
| 105| 4| Chair| Furniture| 250.0| 4|2024-02-01|
+-------+------+-------+-----------+------+--------+----------+
only showing top 5 rows
// 1. Current partitions
println(s"Default partitions: ${employeesDF.rdd.getNumPartitions}")
// 2. Repartition to 4 (full shuffle)
val repartitioned = employeesDF.repartition(4)
println(s"After repartition(4): ${repartitioned.rdd.getNumPartitions}")
// 3. Coalesce to 2 (no full shuffle)
val coalesced = repartitioned.coalesce(2)
println(s"After coalesce(2): ${coalesced.rdd.getNumPartitions}")
// 4. Repartition by column
val byDept = employeesDF.repartition($"department")
println(s"After repartition(dept): ${byDept.rdd.getNumPartitions}")
// 5. Cache
val cachedDF = employeesDF.cache()
cachedDF.count() // trigger materialization
println("DataFrame cached")
// 6. Unpersist
cachedDF.unpersist()
println("DataFrame unpersisted")Default partitions: 8
After repartition(4): 4
After coalesce(2): 2
After repartition(dept): 1
DataFrame cached
DataFrame unpersisted
// 1. countDistinct per category
println("=== countDistinct ===")
salesDF.groupBy("category")
.agg(countDistinct("product").as("distinct_products"))
.show()
// 2. approx_count_distinct
println("=== approx_count_distinct ===")
salesDF.groupBy("category")
.agg(approx_count_distinct("product").as("approx_distinct"))
.show()
// 3. collect_list
println("=== collect_list ===")
salesDF.groupBy("emp_id")
.agg(collect_list("product").as("all_products"))
.show(false)
// 4. collect_set
println("=== collect_set ===")
salesDF.groupBy("emp_id")
.agg(collect_set("product").as("unique_products"))
.show(false)
// 5. first / last by sale_date
val salesOrdered = salesDF.withColumn("sale_date", to_date($"sale_date"))
val empSalesWindow = Window.partitionBy("emp_id").orderBy("sale_date")
println("=== First & Last Sale ===")
salesOrdered.groupBy("emp_id")
.agg(
first("product").as("first_product"),
last("product").as("last_product")
).show()=== countDistinct ===
+-----------+-----------------+
| category|distinct_products|
+-----------+-----------------+
|Electronics| 4|
| Furniture| 2|
+-----------+-----------------+
=== approx_count_distinct ===
+-----------+---------------+
| category|approx_distinct|
+-----------+---------------+
|Electronics| 4|
| Furniture| 2|
+-----------+---------------+
=== collect_list ===
+------+---------------------+
|emp_id|all_products |
+------+---------------------+
|1 |[Laptop, Desk, Phone]|
|2 |[Phone, Monitor] |
|3 |[Laptop] |
|5 |[Laptop] |
|4 |[Chair] |
|6 |[Keyboard] |
|7 |[Desk] |
|9 |[Monitor] |
|8 |[Chair] |
+------+---------------------+
=== collect_set ===
+------+---------------------+
|emp_id|unique_products |
+------+---------------------+
|1 |[Laptop, Phone, Desk]|
|2 |[Phone, Monitor] |
|3 |[Laptop] |
|5 |[Laptop] |
|4 |[Chair] |
|6 |[Keyboard] |
|7 |[Desk] |
|9 |[Monitor] |
|8 |[Chair] |
+------+---------------------+
=== First & Last Sale ===
+------+-------------+------------+
|emp_id|first_product|last_product|
+------+-------------+------------+
| 1| Laptop| Phone|
| 2| Phone| Monitor|
| 3| Laptop| Laptop|
| 4| Chair| Chair|
| 5| Laptop| Laptop|
| 6| Keyboard| Keyboard|
| 7| Desk| Desk|
| 8| Chair| Chair|
| 9| Monitor| Monitor|
+------+-------------+------------+
val deptWindow = Window.partitionBy("department").orderBy($"salary".desc_nulls_last)
// Revenue per employee
val revenueDF = salesDF
.withColumn("revenue", $"price" * $"quantity")
.groupBy("emp_id")
.agg(round(sum("revenue"), 2).as("revenue_per_emp"))
val pipeline = employeesDF
.join(revenueDF, Seq("emp_id"), "left")
.na.fill(Map("revenue_per_emp" -> 0.0))
.join(departmentsDF, employeesDF("department") === departmentsDF("dept_name"), "left")
.withColumn("salary_pct_of_budget",
round(($"salary" / $"budget") * 100, 2))
.withColumn("dept_salary_rank",
rank().over(deptWindow))
.withColumn("perf_label",
when($"revenue_per_emp" > 3000, "Star")
.when($"revenue_per_emp" > 1000, "Good")
.otherwise("Low")
)
.select("name", "department", "salary", "revenue_per_emp",
"salary_pct_of_budget", "dept_salary_rank", "perf_label")
.orderBy("department", "dept_salary_rank")
pipeline.show()+-------+-----------+--------+---------------+--------------------+----------------+----------+
| name| department| salary|revenue_per_emp|salary_pct_of_budget|dept_salary_rank|perf_label|
+-------+-----------+--------+---------------+--------------------+----------------+----------+
| Eve|Engineering|120000.0| 6000.0| 24.0| 1| Star|
|Charlie|Engineering|110000.0| 1200.0| 22.0| 2| Good|
| Hank|Engineering| 98000.0| 1500.0| 19.6| 3| Good|
| Alice|Engineering| 95000.0| 4550.0| 19.0| 4| Star|
| Diana| HR| 60000.0| 1000.0| 40.0| 1| Low|
| Grace| HR| 58000.0| 450.0| 38.67| 2| Low|
| Jack| HR| NULL| 0.0| NULL| 3| Low|
| Ivy| Marketing| 75000.0| 1800.0| 25.0| 1| Good|
| Bob| Marketing| 72000.0| 2000.0| 24.0| 2| Good|
| Frank| Marketing| 68000.0| 800.0| 22.67| 3| Low|
+-------+-----------+--------+---------------+--------------------+----------------+----------+
// 1. Dept with highest avg salary
println("=== Highest Avg Salary Dept ===")
spark.sql("""
SELECT department, ROUND(AVG(salary), 2) AS avg_salary
FROM employees
GROUP BY department
ORDER BY avg_salary DESC
LIMIT 1
""").show()
// 2. Employees earning more than dept avg
println("=== Above Dept Average ===")
spark.sql("""
SELECT e.name, e.department, e.salary, ROUND(d.avg_sal, 2) AS dept_avg
FROM employees e
JOIN (
SELECT department, AVG(salary) AS avg_sal
FROM employees
GROUP BY department
) d ON e.department = d.department
WHERE e.salary > d.avg_sal
""").show()
// 3. Top 2 per dept using ROW_NUMBER
println("=== Top 2 per Dept (SQL Window) ===")
spark.sql("""
SELECT name, department, salary, rn FROM (
SELECT name, department, salary,
ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) AS rn
FROM employees
) ranked
WHERE rn <= 2
""").show()
// 4. Total revenue per department (3-table join)
println("=== Revenue per Dept ===")
spark.sql("""
SELECT e.department, d.dept_head,
ROUND(SUM(s.price * s.quantity), 2) AS total_revenue
FROM employees e
JOIN sales s ON e.emp_id = s.emp_id
JOIN departments d ON e.department = d.dept_name
GROUP BY e.department, d.dept_head
ORDER BY total_revenue DESC
""").show()
// 5. CTE — depts where total salary > 50% of budget
println("=== Salary > 50% of Budget (CTE) ===")
spark.sql("""
WITH dept_salary AS (
SELECT e.department, SUM(e.salary) AS total_salary, d.budget
FROM employees e
JOIN departments d ON e.department = d.dept_name
GROUP BY e.department, d.budget
)
SELECT department,
ROUND(total_salary, 0) AS total_salary,
budget,
ROUND(total_salary/budget*100, 1) AS pct_of_budget
FROM dept_salary
WHERE total_salary > 0.5 * budget
ORDER BY pct_of_budget DESC
""").show()=== Highest Avg Salary Dept ===
+-----------+----------+
| department|avg_salary|
+-----------+----------+
|Engineering| 105750.0|
+-----------+----------+
=== Above Dept Average ===
+-------+-----------+--------+--------+
| name| department| salary|dept_avg|
+-------+-----------+--------+--------+
| Bob| Marketing| 72000.0|71666.67|
|Charlie|Engineering|110000.0|105750.0|
| Diana| HR| 60000.0| 59000.0|
| Eve|Engineering|120000.0|105750.0|
| Ivy| Marketing| 75000.0|71666.67|
+-------+-----------+--------+--------+
=== Top 2 per Dept (SQL Window) ===
+-------+-----------+--------+---+
| name| department| salary| rn|
+-------+-----------+--------+---+
| Eve|Engineering|120000.0| 1|
|Charlie|Engineering|110000.0| 2|
| Diana| HR| 60000.0| 1|
| Grace| HR| 58000.0| 2|
| Ivy| Marketing| 75000.0| 1|
| Bob| Marketing| 72000.0| 2|
+-------+-----------+--------+---+
=== Revenue per Dept ===
+-----------+-------------+-------------+
| department| dept_head|total_revenue|
+-----------+-------------+-------------+
|Engineering|Alice Johnson| 13250.0|
| Marketing| Bob Williams| 4600.0|
| HR| Carol Smith| 1450.0|
+-----------+-------------+-------------+
=== Salary > 50% of Budget (CTE) ===
+-----------+------------+--------+-------------+
| department|total_salary| budget|pct_of_budget|
+-----------+------------+--------+-------------+
|Engineering| 423000.0|500000.0| 84.6|
| HR| 118000.0|150000.0| 78.7|
| Marketing| 215000.0|300000.0| 71.7|
+-----------+------------+--------+-------------+