Case Study Description
Designed and optimized a financial data processing pipeline in Microsoft Fabric using PySpark and Delta Lake. The project focused on improving pipeline performance, enforcing data quality, enabling data recovery, and building a scalable architecture for enterprise-scale financial reporting.
Business Problem
A multinational finance division relied on a daily Accounts Receivable (AR) reporting pipeline that processed 45 million transaction records by joining them with a 15,000-record customer master table.
The existing pipeline faced several challenges:
-
Processing time exceeded 2 hours
-
Large data shuffling caused poor Spark performance
-
Invalid data was loaded because schema validation was missing
-
Corrupted data reached executive dashboards
-
No mechanism existed to recover previous versions after bad data loads
-
Growing data volumes reduced query performance
-
Manual recovery increased operational risk and compliance concerns
Key Objectives
The project aimed to:
-
Reduce pipeline execution time
-
Optimize PySpark joins for large datasets
-
Improve data quality through schema validation
-
Enable version control and data recovery
-
Build a scalable and maintainable data pipeline
-
Improve query performance for analytical reporting
-
Support governance, auditing, and compliance requirements
Solution Architecture
1. Source Files
Raw financial data is collected from multiple sources, including transaction records, customer master data, and reference datasets, to support downstream processing.
2. Data Validation (Schema Enforcement)
Incoming data is validated against predefined schemas to ensure only accurate and properly formatted records enter the pipeline.
3. PySpark Transformations
Data is cleaned, transformed, and prepared using PySpark to create a structured dataset for analysis and reporting.
4. Broadcast Join Optimization
The customer master table is broadcast to all worker nodes, reducing data shuffling and significantly improving join performance.
5. Delta Lake Tables
Processed data is stored in Delta Lake tables, providing ACID transactions, schema enforcement, and reliable version control.
6. Time Travel & RESTORE
Delta Lake's version history is used to view previous table states and quickly restore data if an incorrect or failed load occurs.
7. OPTIMIZE & Z-ORDER
Data files are compacted and reorganized to improve storage efficiency and speed up queries on large datasets.
8. Business Reporting & Analytics
The optimized and trusted data is delivered to business users for accurate reporting, dashboard creation, and data-driven decision-making
1. Source Files
Steps:
Step 1: Define Schema & Load Data
1. Source Files
2. Schema Enforcement
What are we doing?
In this step, we define the schema for the AR Transactions and Customer Master datasets before loading them into PySpark DataFrames. Instead of allowing Spark to automatically detect the data types, we explicitly specify the structure of each dataset to ensure consistent and accurate data ingestion.
Why are we doing this?
Defining the schema improves data quality by preventing incorrect data types, avoids errors caused by automatic schema inference, and speeds up data loading since Spark doesn't need to examine the entire dataset to determine column types. This creates a reliable foundation for all downstream transformations and analysis.
Functions Used
StructType(): Creates the overall schema for a DataFrame by grouping all column definitions together.
StructField(): Defines the properties of an individual column, including its name, data type, and whether it can contain null values.
schema=: Applies the predefined schema while reading the file, ensuring the correct data types are assigned to each column.
inferSchema=True:Â Automatically examines the data and determines the appropriate data type for each column (such as Integer, String, Date, or Float), eliminating the need to define the schema manually.
Step 2: Broadcast Join Optimization
Functions Used
spark.conf.set(): Sets Spark configuration properties during runtime.
spark.sql.autoBroadcastJoinThreshold: Specifies the maximum table size that Spark can automatically broadcast during a join.
join():Â Combines two DataFrames using a common key (CustomerID) with a Left Join.
explain(mode="formatted"):Â Displays Spark's physical execution plan, helping verify the join strategy and identify performance optimizations.
count(): Executes the DataFrame transformations and returns the total number of records.
Execution Plan Highlights
-
AdaptiveSparkPlan – Spark dynamically optimizes the query execution at runtime.
-
BroadcastHashJoin – Confirms that Spark selected the optimized Broadcast Join strategy.
-
BroadcastExchange – Indicates that the Customer Master table was broadcast to all executors before performing the join.
-
Scan CSV – Reads the source CSV files before applying transformations and joins.
What are we doing?
We set Spark's broadcast size limit to 10 MB so it automatically copies small tables across the cluster instead of shuffling all data. Next, we joined the AR Transactions and Customer Master datasets using the CustomerID column. Finally, we checked Spark's execution plan to confirm it successfully used the fast Broadcast Join strategy.
Why are we doing this?
The Customer Master table is much smaller than the AR Transactions table. Broadcasting the smaller table to all worker nodes eliminates expensive data shuffling, resulting in faster join execution and better performance. Reviewing the execution plan confirms that Spark is using the most efficient join strategy.
Broadcast Join: Broadcast Join is a join optimization technique where Spark copies the smaller table to all worker nodes and joins it with the larger table locally. This eliminates data shuffling, making the join much faster for large datasets.
Sort Merge Join: Sort Merge Join is a join strategy where Spark shuffles and sorts both tables based on the join key before merging them. It is commonly used when both tables are large and cannot be broadcast.
In simple terms:
-
Broadcast Join: Best for one small table and one large table → Faster.
-
Sort Merge Join: Best for two large tables → Slower due to sorting and shuffling.
Step 3: Validate Schema Enforcement in Delta Lake
What are we doing?
In this step, we first create a Delta table from the valid Customer Master dataset. Next, we load another dataset that contains an invalid schema (where the Credit_Limit column has an incorrect data type) and try to append it to the Delta table.
Why are we doing this?
The objective is to demonstrate Delta Lake's Schema Enforcement feature. Delta Lake validates the schema of incoming data before writing it to the table. If the data types do not match, it rejects the write operation, preventing invalid or inconsistent data from being stored.
What are we doing?
We update all records in the customer_master_testing Delta table by setting the Credit_Limit to 1. After the update, we check the modified data, view the table's version history, retrieve the original data using Time Travel, and finally restore the table to its previous version.
Why are we doing this?
This demonstrates Delta Lake Time Travel, which allows us to access and restore previous versions of a table. It helps recover data after accidental updates or deletions, supports auditing, and enables tracking of historical changes without requiring backups.
Functions Used
UPDATE: Modifies existing records in the Delta table.
SELECT:Â Displays the current contents of the table after the update.
DESCRIBE HISTORY:Â Shows the version history and all operations performed on the Delta table.
VERSION AS OF:Â Reads data from a specific historical version of the table.
RESTORE TABLE ... VERSION AS OF:Â Restores the Delta table to a selected previous version.