Microsoft Fabric
PySpark

Workforce Intelligence: Azure SQL to Lakehouse

Retiring a manual SQL export and an Excel cleaning ritual that cost HR two days of reporting latency.

Role: Independent project
Year: 2026
Stack: Microsoft Fabric · Azure SQL · PySpark · Lakehouse · Delta Lake

This project implements 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 emphasises data quality, auditability, incremental processing, and reusable transformation logic.

The 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

Two-stage pipeline diagram: Azure SQL dbo.employees flows via an incremental copy activity keyed on updated_at into a bronze_employees table, then a PySpark cleanse notebook reads departments.csv and writes to silver_employees, with a dashed path to silver_quarantine. — Bronze holds the raw landing copy; Silver holds only records that earned their place. Everything rejected goes to a quarantine table rather than being dropped — nothing disappears silently.

Data quality challenges

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

Issue	Resolution
Null hire dates	Quarantined
Null salaries	Quarantined
Invalid salary values	Quarantined
Duplicate employee IDs	Retained latest record using `updated_at`
Name prefixes	Extracted into a separate column
Missing analytics layer	Bronze–Silver architecture

Quarantining rather than deleting is the decision that matters here. A dropped row is a number nobody can reconcile later; a quarantined row is a number someone can go and fix.

The transformation

Inspect — profile the raw Bronze table and establish row counts before any change.
Quarantine — filter out null hire dates, null salaries, and invalid salary values, tagging each with its rejection reason and unioning them into a quarantine table.
Deduplicate — order by updated_at descending and drop duplicates on employee_id, keeping the most recent record for each employee.
Enrich, join, stamp, and write — extract name prefixes, derive hire year, tenure and salary band, join the department reference data, stamp audit columns, and write the governed Silver table.
Reconcile — count rows across every stage so Bronze, Silver, and quarantine always add up.

The enrichment step does the bulk of the work in a single chained expression:

working_df1 = working_df0.withColumn("name_prefix", F.when(F.col("employee_name").startswith("Dr."), "Dr.")
                    .when(F.col("employee_name").startswith("Mr."), "Mr.")
                    .when(F.col("employee_name").startswith("Mrs."), "Mrs.")
                    .when(F.col("employee_name").startswith("Ms."), "Ms.")
                    .otherwise("")) \
    .withColumn("employee_name", F.trim(F.expr("substring(employee_name, len(name_prefix)+1)"))) \
    .withColumn("hire_year", F.year("hire_date")) \
    .withColumn("tenure_years", F.round(F.datediff(F.current_date(), F.col("hire_date")) / 365.25, 2)) \
    .withColumn("salary_band", F.when(F.col("salary") >= 120000, "Senior")
                    .when(F.col("salary") >= 90000, "Mid").otherwise("Junior")) \
    .join(dept_df, "department_id", "left") \
    .withColumn("transformed_at", F.current_timestamp()) \
    .withColumn("employee_source", F.lit("Azure_SQL")) \
    .withColumn("dept_source", F.lit("Lakehouse_CSV_File"))

The employee_source and dept_source literals are deliberate: six months later, when a number looks wrong, the Silver table itself says which system it came from.

Fabric notebook cell running the enrichment chain, reporting 5 of 5 Spark jobs succeeded, with a results table of 17 columns and 307 rows showing department_id, employee_id, employee_name, job_title, hire_date, salary, updated_at, name_prefix, hire_year and tenure columns. — The enrichment cell executing in the Fabric notebook — 17 columns out, with prefix, hire year and tenure derived in the same pass as the department join.

PySpark functions used in the cleanse and enrich stages22 functions

spark.read.csv(): Loads raw CSV data into a PySpark DataFrame, parsing structural parameters like delimiters and headers.
spark.read.table(): Loads data directly from a registered Lakehouse Delta table into a DataFrame.
display(): A native Fabric utility that renders DataFrames as interactive visual tables for quick profiling.
.filter() / F.col(): Filters rows against a condition; F.col() references a column by name to apply expressions or transformations.
.isNull() / .isNotNull(): Evaluate column values for missing or present data — the basis of the quarantine rules.
| and & (bitwise OR / AND): Combine multiple filter conditions: OR to catch any rejection reason, AND to require every validity criterion.
.withColumn(): Creates a new column or replaces an existing one by applying a transformation or assigning a value.
F.lit(): Creates a literal constant column, appending static text or numbers to every row.
F.when().otherwise(): Implements IF-ELSE conditional logic to assign values based on specific criteria.
.union(): Combines two DataFrames vertically, requiring both to share the same schema — used to assemble the quarantine table.
.orderBy(…, .desc()) / .dropDuplicates(): Sort descending then drop duplicates on employee_id, retaining the most recent record per employee.
F.trim() / F.expr("substring(...)"): Remove surrounding whitespace and extract portions of text via a SQL-style expression.
F.year() / F.datediff() / F.current_date() / F.current_timestamp(): Date components and arithmetic used to derive hire year, tenure, and audit timestamps.
F.round(): Rounds a numeric value to a specified number of decimal places.
.join(..., "left"): Left-outer joins the department reference data, keeping every employee record even where no department matches.
.write / .mode("overwrite") / .saveAsTable(): Persists the DataFrame into the Lakehouse metastore as a managed Delta table, replacing any existing data and schema.
.count() / print(f"..."): Counts rows and logs the totals — the reconciliation step that proves no records were lost between stages.

Analytics challenges

Three HR questions that each need a window function rather than a group-by.

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: the same salary should produce the same rank.

W.partitionBy() defines the window boundary so the ranking is calculated independently within each department, .orderBy(F.col("salary").desc()) sorts highest to lowest, and F.rank().over(window_spec) assigns the rank — tied values receive the same rank and the sequence skips accordingly (1, 2, 2, 4).

Salary versus department average

HR wants to understand how each employee’s salary compares to the average in their department. Add a column showing the difference. Positive means above average, negative means below.

F.round(F.col("salary") - F.avg("salary").over(w_agg), 0) computes the variance against the partitioned average. To validate the result, the data is grouped by department to count how many salaries fall above and below the department average alongside total headcount — if those counts don’t sum to the headcount, the window is wrong.

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("region").orderBy(F.col("salary").desc()) groups and sorts within each region, F.row_number() assigns a unique sequential number so the top earner always receives 1, and .filter(F.col("row_num") == 1) isolates them. Because the partition is derived from the data, a new region appears in the output automatically.