amee-joshi-data-engineering-portfolio

Skill by ara.so — Data Skills collection.

This portfolio showcases production-grade data engineering patterns and architectures for building scalable, cloud-native data platforms. It demonstrates end-to-end solutions covering data ingestion, transformation, modeling, and analytics using Azure services, Databricks, SQL Server, and BI tools.

What This Portfolio Demonstrates

This is a reference collection showing:

• Medallion Architecture (Bronze-Silver-Gold) implementations
• Azure cloud data platforms (ADF, ADLS Gen2, Databricks, Synapse Analytics)
• Data lakehouse patterns with Delta Lake
• Dimensional modeling (Star Schema, SCD Type 1 & 2)
• Metadata-driven ingestion frameworks
• Analytics-ready datasets for BI consumption
• ETL/ELT pipeline design with incremental loading
• Power BI and Tableau reporting solutions

Key Portfolio Projects

1. Azure Databricks Retail Lakehouse

Repository: azure-databricks-end-to-end-retail-lakehouse

Pattern: Enterprise Medallion Architecture with Delta Lake

Architecture:

Bronze (Raw) → Silver (Cleansed) → Gold (Analytics-Ready)

Key Implementation Concepts:

# Bronze Layer - Raw Ingestionfrom pyspark.sql import SparkSessionfrom delta.tables import DeltaTable# Ingest raw data with metadatadf_raw = (spark.read    .format("parquet")    .load(f"{bronze_path}/source_data/")    .withColumn("ingestion_timestamp", current_timestamp())    .withColumn("source_file", input_file_name()))# Write to Bronze Delta table(df_raw.write    .format("delta")    .mode("append")    .option("mergeSchema", "true")    .save(f"{bronze_path}/retail_transactions"))

# Silver Layer - Data Quality & Transformationfrom pyspark.sql.functions import col, when, trim, upper# Cleanse and standardizedf_silver = (df_bronze    .filter(col("transaction_id").isNotNull())    .withColumn("customer_name", trim(upper(col("customer_name"))))    .withColumn("transaction_amount",                 when(col("transaction_amount") < 0, 0)                .otherwise(col("transaction_amount")))    .dropDuplicates(["transaction_id"])    .select("transaction_id", "customer_id", "product_id",             "transaction_amount", "transaction_date"))# Write with schema enforcement(df_silver.write    .format("delta")    .mode("overwrite")    .option("overwriteSchema", "false")    .save(f"{silver_path}/transactions"))

# Gold Layer - SCD Type 2 Dimensiondef apply_scd_type2(target_table, source_df, key_columns, scd_columns):    """    Implements Slowly Changing Dimension Type 2    """    from delta.tables import DeltaTable    from pyspark.sql.functions import lit, current_timestamp        # Prepare source with SCD metadata    source_prepared = (source_df        .withColumn("effective_date", current_timestamp())        .withColumn("end_date", lit(None).cast("timestamp"))        .withColumn("is_current", lit(True))    )        # Read existing target    target_delta = DeltaTable.forPath(spark, target_table)        # Identify changes    merge_condition = " AND ".join([f"target.{k} = source.{k}" for k in key_columns])        # Perform SCD Type 2 merge    (target_delta.alias("target")        .merge(source_prepared.alias("source"), merge_condition)        .whenMatchedUpdate(            condition = "target.is_current = true AND " +                        " OR ".join([f"target.{c} != source.{c}" for c in scd_columns]),            set = {                "is_current": "false",                "end_date": "current_timestamp()"            }        )        .whenNotMatchedInsertAll()        .execute()    )

2. Metadata-Driven Ingestion Framework

Pattern: Dynamic, configuration-based pipeline generation

Configuration Schema:

{  "pipeline_config": {    "source_system": "SQL_SERVER",    "target_layer": "bronze",    "ingestion_type": "incremental",    "watermark_column": "modified_date",    "tables": [      {        "schema_name": "sales",        "table_name": "orders",        "partition_column": "order_date",        "primary_key": ["order_id"],        "target_path": "/bronze/sales/orders"      }    ]  }}

Azure Data Factory Pattern:

# Dynamic pipeline parameter processing# This represents the logic implemented in ADFdef generate_copy_activity(table_config):    """    Generates ADF copy activity from metadata    """    return {        "name": f"Copy_{table_config['table_name']}",        "type": "Copy",        "inputs": [{            "referenceName": "SourceDataset",            "type": "DatasetReference",            "parameters": {                "schemaName": table_config['schema_name'],                "tableName": table_config['table_name']            }        }],        "outputs": [{            "referenceName": "SinkDataset",            "type": "DatasetReference",            "parameters": {                "targetPath": table_config['target_path']            }        }],        "typeProperties": {            "source": {                "type": "SqlServerSource",                "sqlReaderQuery": f"""                    SELECT * FROM {table_config['schema_name']}.{table_config['table_name']}                    WHERE {table_config['watermark_column']} > '@{{pipeline().parameters.watermarkValue}}'                """            },            "sink": {                "type": "ParquetSink",                "storeSettings": {                    "type": "AzureBlobFSWriteSettings",                    "copyBehavior": "PreserveHierarchy"                }            }        }    }

3. Star Schema Data Warehouse

Pattern: Dimensional Modeling with SQL Server

Dimension Table (SCD Type 1):

-- Dimension: Product (SCD Type 1)CREATE TABLE dim_product (    product_key INT IDENTITY(1,1) PRIMARY KEY,    product_id INT NOT NULL,    product_name NVARCHAR(100),    category NVARCHAR(50),    subcategory NVARCHAR(50),    unit_price DECIMAL(10,2),    modified_date DATETIME DEFAULT GETDATE(),    CONSTRAINT uk_product UNIQUE (product_id));-- ETL Merge (SCD Type 1 - Overwrite)MERGE INTO dim_product AS targetUSING (    SELECT         product_id,        product_name,        category,        subcategory,        unit_price    FROM staging.products) AS sourceON target.product_id = source.product_idWHEN MATCHED AND (    target.product_name != source.product_name OR    target.category != source.category OR    target.unit_price != source.unit_price)THEN UPDATE SET    target.product_name = source.product_name,    target.category = source.category,    target.subcategory = source.subcategory,    target.unit_price = source.unit_price,    target.modified_date = GETDATE()WHEN NOT MATCHED BY TARGETTHEN INSERT (product_id, product_name, category, subcategory, unit_price)VALUES (source.product_id, source.product_name, source.category,         source.subcategory, source.unit_price);

Dimension Table (SCD Type 2):

-- Dimension: Customer (SCD Type 2)CREATE TABLE dim_customer (    customer_key INT IDENTITY(1,1) PRIMARY KEY,    customer_id INT NOT NULL,    customer_name NVARCHAR(100),    email NVARCHAR(100),    city NVARCHAR(50),    state NVARCHAR(50),    effective_date DATETIME NOT NULL,    end_date DATETIME NULL,    is_current BIT DEFAULT 1,    CONSTRAINT uk_customer_current UNIQUE (customer_id, is_current));-- ETL for SCD Type 2-- Step 1: Expire changed recordsUPDATE dim_customerSET     end_date = GETDATE(),    is_current = 0WHERE customer_id IN (    SELECT s.customer_id    FROM staging.customers s    INNER JOIN dim_customer d ON s.customer_id = d.customer_id    WHERE d.is_current = 1    AND (s.city != d.city OR s.state != d.state));-- Step 2: Insert new versionsINSERT INTO dim_customer (    customer_id, customer_name, email, city, state,     effective_date, end_date, is_current)SELECT     s.customer_id,    s.customer_name,    s.email,    s.city,    s.state,    GETDATE() AS effective_date,    NULL AS end_date,    1 AS is_currentFROM staging.customers sLEFT JOIN dim_customer d ON s.customer_id = d.customer_id AND d.is_current = 1WHERE d.customer_key IS NULL   OR s.city != d.city   OR s.state != d.state;

Fact Table:

-- Fact: Sales TransactionsCREATE TABLE fact_sales (    sales_key BIGINT IDENTITY(1,1) PRIMARY KEY,    date_key INT NOT NULL,    customer_key INT NOT NULL,    product_key INT NOT NULL,    store_key INT NOT NULL,    quantity INT NOT NULL,    unit_price DECIMAL(10,2) NOT NULL,    discount_amount DECIMAL(10,2) DEFAULT 0,    tax_amount DECIMAL(10,2) DEFAULT 0,    total_amount DECIMAL(10,2) NOT NULL,    CONSTRAINT fk_date FOREIGN KEY (date_key) REFERENCES dim_date(date_key),    CONSTRAINT fk_customer FOREIGN KEY (customer_key) REFERENCES dim_customer(customer_key),    CONSTRAINT fk_product FOREIGN KEY (product_key) REFERENCES dim_product(product_key),    CONSTRAINT fk_store FOREIGN KEY (store_key) REFERENCES dim_store(store_key));-- Create columnstore index for analyticsCREATE NONCLUSTERED COLUMNSTORE INDEX idx_fact_sales_csON fact_sales (date_key, customer_key, product_key, store_key,                quantity, unit_price, total_amount);-- ETL LoadINSERT INTO fact_sales (    date_key, customer_key, product_key, store_key,    quantity, unit_price, discount_amount, tax_amount, total_amount)SELECT     dd.date_key,    dc.customer_key,    dp.product_key,    ds.store_key,    st.quantity,    st.unit_price,    st.discount_amount,    st.tax_amount,    st.total_amountFROM staging.transactions stINNER JOIN dim_date dd ON CAST(st.transaction_date AS DATE) = dd.dateINNER JOIN dim_customer dc ON st.customer_id = dc.customer_id AND dc.is_current = 1INNER JOIN dim_product dp ON st.product_id = dp.product_idINNER JOIN dim_store ds ON st.store_id = ds.store_id;

4. Incremental Data Loading Pattern

Watermark-Based Incremental Load:

# Databricks notebook - Incremental load with watermarkfrom pyspark.sql.functions import col, max as spark_maxfrom delta.tables import DeltaTable# Configurationsource_table = "source_database.transactions"target_path = "/mnt/silver/transactions"watermark_table = "control.watermark"watermark_column = "modified_date"# Get last watermarklast_watermark = (spark.table(watermark_table)    .filter(col("table_name") == source_table)    .select("watermark_value")    .first()[0])# Read incremental datadf_incremental = (spark.table(source_table)    .filter(col(watermark_column) > last_watermark))# Check if target existsif DeltaTable.isDeltaTable(spark, target_path):    # Merge into existing table    target_table = DeltaTable.forPath(spark, target_path)        (target_table.alias("target")        .merge(            df_incremental.alias("source"),            "target.transaction_id = source.transaction_id"        )        .whenMatchedUpdateAll()        .whenNotMatchedInsertAll()        .execute()    )else:    # Initial load    (df_incremental.write        .format("delta")        .mode("overwrite")        .save(target_path)    )# Update watermarknew_watermark = df_incremental.agg(spark_max(watermark_column)).first()[0]spark.sql(f"""    UPDATE {watermark_table}    SET watermark_value = '{new_watermark}',        last_updated = current_timestamp()    WHERE table_name = '{source_table}'""")

5. Data Quality Framework

Quality Checks Pattern:

from pyspark.sql.functions import col, count, sum as spark_sum, whenclass DataQualityChecker:    """    Data quality validation framework    """        def __init__(self, dataframe, table_name):        self.df = dataframe        self.table_name = table_name        self.quality_results = []        def check_null_values(self, columns):        """Check for null values in critical columns"""        for column in columns:            null_count = self.df.filter(col(column).isNull()).count()            total_count = self.df.count()                        self.quality_results.append({                "check_type": "null_check",                "column": column,                "null_count": null_count,                "total_count": total_count,                "null_percentage": (null_count / total_count * 100) if total_count > 0 else 0,                "passed": null_count == 0            })        return self        def check_duplicates(self, key_columns):        """Check for duplicate records"""        duplicate_count = (self.df            .groupBy(key_columns)            .count()            .filter(col("count") > 1)            .count()        )                self.quality_results.append({            "check_type": "duplicate_check",            "key_columns": key_columns,            "duplicate_count": duplicate_count,            "passed": duplicate_count == 0        })        return self        def check_referential_integrity(self, foreign_key, reference_df, reference_key):        """Check referential integrity"""        missing_references = (self.df            .select(foreign_key)            .distinct()            .join(reference_df.select(reference_key),                   col(foreign_key) == col(reference_key),                   "left_anti")            .count()        )                self.quality_results.append({            "check_type": "referential_integrity",            "foreign_key": foreign_key,            "missing_references": missing_references,            "passed": missing_references == 0        })        return self        def check_value_range(self, column, min_value=None, max_value=None):        """Check if values are within expected range"""        out_of_range = self.df.filter(            (col(column) < min_value if min_value is not None else False) |            (col(column) > max_value if max_value is not None else False)        ).count()                self.quality_results.append({            "check_type": "range_check",            "column": column,            "min_value": min_value,            "max_value": max_value,            "out_of_range_count": out_of_range,            "passed": out_of_range == 0        })        return self        def get_results(self):        """Return quality check results"""        return self.quality_results# Usage exampledf_transactions = spark.read.format("delta").load("/mnt/silver/transactions")df_customers = spark.read.format("delta").load("/mnt/gold/dim_customer")quality_checker = DataQualityChecker(df_transactions, "transactions")results = (quality_checker    .check_null_values(["transaction_id", "customer_id", "transaction_date"])    .check_duplicates(["transaction_id"])    .check_referential_integrity("customer_id", df_customers, "customer_id")    .check_value_range("transaction_amount", min_value=0, max_value=100000)    .get_results())# Log resultsfor result in results:    print(f"{result['check_type']}: {'PASSED' if result['passed'] else 'FAILED'}")

Power BI Analytics Patterns

DAX Measures for KPIs:

// Total SalesTotal Sales = SUM(fact_sales[total_amount])// Year-over-Year GrowthSales YoY Growth = VAR CurrentYearSales = [Total Sales]VAR PreviousYearSales =     CALCULATE(        [Total Sales],        DATEADD(dim_date[Date], -1, YEAR)    )RETURN    DIVIDE(        CurrentYearSales - PreviousYearSales,        PreviousYearSales,        0    )// Customer Lifetime ValueCustomer LTV = CALCULATE(    [Total Sales],    ALLEXCEPT(dim_customer, dim_customer[customer_id]))// Moving Average (3 months)Sales 3M MA = CALCULATE(    [Total Sales],    DATESINPERIOD(        dim_date[Date],        LASTDATE(dim_date[Date]),        -3,        MONTH    )) / 3// Rank by SalesProduct Sales Rank = RANKX(    ALL(dim_product[product_name]),    [Total Sales],    ,    DESC,    DENSE)

Common Architectural Patterns

Medallion Architecture Best Practices

Bronze Layer:

• Raw data ingestion with minimal transformation
• Add audit columns (ingestion_timestamp, source_file)
• Preserve source schema with schema evolution enabled
• Partition by ingestion date for performance

Silver Layer:

• Data cleansing and standardization
• Deduplication based on business keys
• Data type conversions and validations
• Enforce schema constraints
• Join related datasets

Gold Layer:

• Business-aggregated datasets
• Dimensional models (Star/Snowflake schema)
• Pre-calculated metrics and KPIs
• Optimized for BI tool consumption

Delta Lake Optimization

# Optimize Delta tablesfrom delta.tables import DeltaTable# Optimize with Z-orderingdeltaTable = DeltaTable.forPath(spark, "/mnt/gold/fact_sales")# Optimize files and Z-order by common filter columnsdeltaTable.optimize().executeZOrderBy("date_key", "customer_key")# Vacuum old files (retention 168 hours = 7 days)deltaTable.vacuum(168)# Update table statisticsspark.sql("ANALYZE TABLE gold.fact_sales COMPUTE STATISTICS FOR ALL COLUMNS")

Unity Catalog Security

-- Create catalog and schemaCREATE CATALOG IF NOT EXISTS retail_analytics;CREATE SCHEMA IF NOT EXISTS retail_analytics.gold;-- Grant permissionsGRANT USE CATALOG ON CATALOG retail_analytics TO `data_analysts`;GRANT USE SCHEMA ON SCHEMA retail_analytics.gold TO `data_analysts`;GRANT SELECT ON TABLE retail_analytics.gold.fact_sales TO `data_analysts`;-- Row-level securityCREATE FUNCTION retail_analytics.gold.customer_filter(customer_region STRING)RETURN customer_region = current_user_region();ALTER TABLE retail_analytics.gold.fact_sales SET ROW FILTER retail_analytics.gold.customer_filter ON (region);

Environment Setup

Azure Configuration:

# Set Azure environment variablesexport AZURE_SUBSCRIPTION_ID=your_subscription_idexport AZURE_RESOURCE_GROUP=rg-data-platformexport AZURE_STORAGE_ACCOUNT=datalakestorageexport AZURE_DATABRICKS_WORKSPACE=databricks-workspace# ADF connectionexport ADF_FACTORY_NAME=adf-data-ingestionexport ADF_LINKED_SERVICE_NAME=ls-sqlserver-source

Databricks Configuration:

# Mount ADLS Gen2 in Databricksconfigs = {    "fs.azure.account.auth.type": "OAuth",    "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",    "fs.azure.account.oauth2.client.id": dbutils.secrets.get(scope="keyvault", key="client-id"),    "fs.azure.account.oauth2.client.secret": dbutils.secrets.get(scope="keyvault", key="client-secret"),    "fs.azure.account.oauth2.client.endpoint": f"https://login.microsoftonline.com/{dbutils.secrets.get(scope='keyvault', key='tenant-id')}/oauth2/token"}dbutils.fs.mount(    source = "abfss://[email protected]/",    mount_point = "/mnt/bronze",    extra_configs = configs)

Troubleshooting

Issue: Delta Lake merge taking too long

# Solution: Optimize before mergefrom delta.tables import DeltaTabletarget_table = DeltaTable.forPath(spark, target_path)# Compact small files firsttarget_table.optimize().executeCompaction()# Enable auto-optimize and auto-compactionspark.sql(f"""    ALTER TABLE delta.`{target_path}`    SET TBLPROPERTIES (        delta.autoOptimize.optimizeWrite = true,        delta.autoOptimize.autoCompact = true    )""")

Issue: ADF pipeline timeout

// Increase timeout in ADF pipeline activity{  "typeProperties": {    "timeout": "0.12:00:00"  },  "policy": {    "timeout": "7.00:00:00",    "retry": 2,    "retryIntervalInSeconds": 30  }}

Issue: Power BI slow refresh

// Use incremental refresh configuration// In Power BI Desktop: Table Tools > Incremental Refresh// Or optimize DAX measuresOptimized Total Sales = CALCULATE(    SUM(fact_sales[total_amount]),    KEEPFILTERS(dim_date[Date])  // Reduce context transition overhead)

Issue: Schema evolution conflicts

# Enable schema merging in Delta writes(df.write    .format("delta")    .mode("append")    .option("mergeSchema", "true")    .save(target_path))# Or explicitly allow schema overwrite(df.write    .format("delta")    .mode("overwrite")    .option("overwriteSchema", "true")    .save(target_path))

Reference Architecture

This portfolio demonstrates a typical enterprise data platform architecture:

┌─────────────────┐│  Source Systems ││  (SQL Server,   ││   APIs, Files)  │└────────┬────────┘         │         ▼┌─────────────────┐│  Azure Data     ││  Factory (ADF)  │ ◄──── Metadata-driven ingestion└────────┬────────┘         │         ▼┌─────────────────┐│  ADLS Gen2      ││  Bronze Layer   │ ◄──── Raw data landing└────────┬────────┘         │         ▼┌─────────────────┐│  Databricks     ││  Silver Layer   │ ◄──── Cleansing & transformation└────────┬────────┘         │         ▼┌─────────────────┐│  Databricks     ││  Gold Layer     │ ◄──── Analytics-ready datasets└────────┬────────┘         │         ├──────────────────┐         ▼                  ▼┌─────────────────┐  ┌─────────────────┐│  Power BI       │  │  Synapse        ││  Reporting      │  │  Analytics      │└─────────────────┘  └─────────────────┘

This skill provides patterns and code examples for building production-grade data platforms following industry best practices demonstrated across the portfolio projects.