Workforce Intelligence.
From Azure SQL
to Lakehouse.

Project Overview

This project demonstrates the implementation of a modern Microsoft Fabric data engineering pipeline that transforms raw workforce data into a trusted analytical dataset for Human Resources reporting.

The solution replaces a manual SQL export and Excel-based reporting workflow with an automated Bronze-to-Silver Lakehouse architecture built using Microsoft Fabric, Azure SQL Database, and PySpark. The pipeline emphasizes data quality, auditability, incremental processing, and reusable transformation logic.

Business Problem

The HR department relied on:

Manual SQL exports
Excel-based cleaning
Duplicate employee records
Invalid salary values
Missing hire dates
No governed analytical layer
Approximately 48-hour reporting delays

The objective was to design a scalable Microsoft Fabric solution capable of producing clean, trusted workforce data for downstream reporting.

Solution Architecture

Data Quality Challenges

The source system contained multiple quality issues that required cleansing before analytics.

Issue Solution

Null Hire Dates Quarantined

Null Salaries Quarantined

Invalid Salary Values Quarantined

Duplicate Employee IDs Retained latest record using updated_at

Name Prefixes Extracted into separate column

Missing Analytics Layer Bronze-Silver Architecture

Bonus Challenges

Bonus 01: Rank
Bonus 02: Compare
Bonus 03: Aggregate

Solution

Step 1

Inspect

Step 2

Quarantine

Step 3

Deduplicate

Step 4-6

Enrich, Join, Stamp & Write

Dataset 1

Dataset 2

Fabric Project

Function Used:

Bronze_Layer

spark.read.csv(): Loads raw CSV data into a PySpark DataFrame by parsing structural parameters like delimiters and headers.

display(): A native Fabric utility that renders DataFrames into interactive visual tables for quick profiling.

spark.read.table(): Loads data directly from a registered Lakehouse delta table into a PySpark DataFrame.

Silver_Quarentine

from pyspark.sql import functions as F: Imports the PySpark SQL functions library, providing access to essential data transformation, aggregation, and manipulation methods used throughout the pipeline.

.filter(): Filters rows in a DataFrame based on specified conditions, retaining only the records that meet the criteria.

F.col(): References a specific column in the DataFrame by its name to apply expressions or transformations.

.isNull(): Evaluates column values and returns true if a record contains a missing or null value.

| (Bitwise OR): Combines multiple filter conditions, returning rows where at least one of the conditions evaluates to true.

.withColumn(): Creates a new column or replaces an existing one by applying a transformation or assigning a value.

F.lit(): Creates a literal or constant value column, allowing static text or numbers to be appended to every row.

.union(): Combines two DataFrames vertically by appending the rows of one DataFrame to another, requiring both to share the same schema structure.

.write: Accesses the DataFrameWriter interface used to save DataFrame contents to external storage or a metastore.

.mode("Overwrite"): Specifies the save behavior, configured to completely replace any existing data and schema in the target destination.

.saveAsTable(): Persists the DataFrame directly into the Lakehouse metastore as a managed Delta table at the specified path.

Silver_Layer

.isNotNull(): Evaluates column values, keeping only those that contain valid, non-null data.

& (Bitwise AND): Combines multiple filter conditions, ensuring only rows meeting all specified criteria remain.

.orderBy(..., .desc()): Sorts the DataFrame by a specific column in descending order, ensuring the most recent entries appear first.

.dropDuplicates(): Identifies and removes duplicate records based on the specified column—in this case, employee_id—retaining only the first (latest) occurrence.

.withColumn(): Adds or replaces a column by applying a transformation expression; overwrites existing columns if the same name is provided.

F.when().otherwise(): Implements conditional logic similar to an IF-ELSE statement to assign values based on specific criteria

F.trim(): Removes leading and trailing whitespace from a string column.

F.expr("substring(...)"): Executes a SQL-style substring expression to extract portions of text data.

F.round(): Rounds a numeric value to a specified number of decimal places.

F.year(): Extracts the year component from a date or timestamp column.

F.datediff(): Calculates the difference in days between two date or timestamp columns.

F.current_date(): Returns the current system date as a date type.

F.current_timestamp(): Returns the current system timestamp.

F.lit(): Creates a column with a constant, static literal value for every row.

.join(..., "left"): Combines two DataFrames based on a shared key column using a left-outer join, keeping all records from the left DataFrame.

Step 7

Reconcile

Bonus Challenges Solutions

Bonus 01: Rank

Salary Rank Within Department
HR wants to know where each employee ranks within their department by salary. Produce a ranking that resets for each department. Handle tied salaries fairly: same salary should produce the same rank.

W.partitionBy(): Defines the window boundary by grouping the data into logical partitions (e.g., by department_id) so that subsequent window functions, like rank(), are calculated independently within each specific group.

.orderBy(F.col("column_name").desc()): Specifies the sorting order within each window partition, arranging records by the target column in descending order (highest to lowest).

F.rank().over(window_spec): Assigns a rank to each row within the defined partition, where tied values receive the same rank and the ranking sequence skips numbers accordingly (e.g., 1, 2, 2, 4).

Bonus 02: Compare

Salary vs Department Average
HR wants to understand how each employee's salary compares to the average in their department. Add a new column showing the difference. A positive value means above average. A negative value means below.

Bonus 03: Aggregate

Top Earner Per Region
The HR Director wants to know who earns the most in each region. Your solution must work regardless of how many regions exist in the data. Do not hardcode the region count.

W.partitionBy(): Defines the window boundary by grouping the data into logical partitions (e.g., by department_name) so that subsequent window functions, like avg(), are calculated independently within each specific group.

F.round((F.col("column_name") - F.avg("column_name").over(w_agg)), 0): Calculates the variance between an individual's salary and their department's average salary, rounding the resulting difference to the nearest whole number.

To validate the analytical results, the data is grouped by department to calculate the frequency of salaries above and below the department average, alongside total headcounts per department.

W.partitionBy("region").orderBy(...): Groups the data by region and sorts salaries in descending order within each group to identify the highest earner.

F.row_number().over(w_spec_reg): Assigns a unique, sequential number to each row within the region, ensuring the top earner always receives a 1.

.filter(F.col("row_num") == 1): Selects only the records where the row number is 1, effectively isolating the top earner for every region.

.select("region", F.col("...").alias("..."), "salary"): Retains only the necessary columns while renaming the employee name column for clear, professional reporting.

.count(): A PySpark action that calculates and returns the total number of rows in a DataFrame, essential for auditing data volumes across pipeline stages.

print(f"..."): A standard Python function utilizing f-strings to dynamically format and display text alongside calculated variable values for quick output logging.

.groupBy("column_name"): Clusters rows by department to enable aggregate calculations for each group.

F.sum(...): Aggregates conditional counts to total how many employees fall above or below the department average.

F.count("*"): Calculates the total number of records within each department grouping.

.alias(...): Renames resulting aggregate columns to provide clear, readable labels in the final table.

Workforce Intelligence. From Azure SQL to Lakehouse.