microsoft-fabric-unified-analytics

# Microsoft Fabric Unified Analytics Platform > Skill by [ara.so](https://ara.so) — Data Skills collection. This project demonstrates how to build a production-grade unified analytics platform using **Microsoft Fabric**, implementing the **Medallion Architecture** (Bronze → Silver → Gold) with **OneLake**, **Dataflow Gen2**, **Fabric Notebooks (PySpark)**, **Semantic Models**, and **Power BI**. ## What This Project Does Microsoft Fabric Unified Analytics Platform showcases: - **Lakehouse Architecture**: Organize data in Bronze (raw), Silver (cleansed), and Gold (business-ready) layers - **Unified SaaS Platform**: Single environment for ingestion, processing, modeling, and visualization - **OneLake Storage**: Centralized data lake foundation for all workloads - **Low-Code + Code**: Combine Dataflow Gen2 (visual) with PySpark notebooks (code) - **Semantic Modeling**: Create reusable business metrics and KPIs - **Native BI Integration**: Power BI directly connected to the platform ## Architecture Overview ``` Raw Data → Bronze Layer (OneLake) ↓ Dataflow Gen2 → Silver Layer (Cleansed) ↓ Fabric Notebooks (PySpark) → Gold Layer (Business KPIs) ↓ Semantic Model → Power BI Reports ``` ## Prerequisites - **Microsoft Fabric Workspace**: With appropriate capacity (F64 or higher recommended) - **Microsoft Fabric License**: Premium or trial capacity - **Power BI Pro/Premium**: For report publishing - **Python 3.8+**: For local notebook development (optional) - **Azure Storage Account**: If ingesting from external sources ## Setting Up Microsoft Fabric Environment ### 1. Create Fabric Workspace ```python # Fabric workspaces are created via the web UI # Navigate to: https://app.fabric.microsoft.com # Click "Workspaces" → "New Workspace" # Name: "RetailAnalyticsPlatform" # Assign Fabric capacity ``` ### 2. Create Lakehouse In your Fabric workspace: 1. Click **New** → **Lakehouse** 2. Name: `retail_analytics_lakehouse` 3. This creates OneLake storage with Delta tables support ### 3. Organize Medallion Layers Create folder structure in your lakehouse: ``` Files/ ├── bronze/ │ ├── online_retail/ │ │ └── online_retail.csv ├── silver/ │ ├── online_retail_cleaned/ └── gold/ ├── revenue_trends/ ├── product_performance/ ├── customer_analytics/ └── rfm_segmentation/ ``` ## Data Ingestion to Bronze Layer ### Upload Raw Data via Lakehouse UI ```python # Files can be uploaded directly via the Lakehouse explorer # Or programmatically using Fabric APIs # Place CSV files in: Files/bronze/online_retail/ ``` ### Using Fabric Notebook for Ingestion ```python # Fabric Notebook - Data Ingestion from pyspark.sql import SparkSession # Fabric provides pre-configured Spark session spark = SparkSession.builder.getOrCreate() # Read from external source (e.g., Azure Blob) storage_account = "your_storage_account" container = "raw-data" sas_token = mssparkutils.credentials.getSecret("KeyVault", "SASToken") df = spark.read.format("csv") \ .option("header", "true") \ .option("inferSchema", "true") \ .load(f"wasbs://{container}@{storage_account}.blob.core.windows.net/online_retail.csv") # Write to Bronze layer as Delta table df.write.format("delta") \ .mode("overwrite") \ .save("Files/bronze/online_retail") print(f"Ingested {df.count()} records to Bronze layer") ``` ## Dataflow Gen2: Bronze to Silver Transformation ### Creating a Dataflow Gen2 1. In Fabric workspace: **New** → **Dataflow Gen2** 2. **Get data** → **Lakehouse** → Select Bronze layer table 3. Apply transformations using Power Query M language ### Key Transformations (Power Query M) ```m let // Load from Bronze Source = Lakehouse.Contents(null), BronzeData = Source{[workspaceId="YOUR_WORKSPACE_ID"]}[Data], // Remove duplicates RemovedDuplicates = Table.Distinct(BronzeData, {"InvoiceNo", "StockCode"}), // Handle missing values RemovedNulls = Table.SelectRows(RemovedDuplicates, each [CustomerID] <> null), // Add business columns AddedLineTotal = Table.AddColumn(RemovedNulls, "line_total", each [Quantity] * [UnitPrice], type number), // Extract date components AddedYear = Table.AddColumn(AddedLineTotal, "year", each Date.Year([InvoiceDate]), Int64.Type), AddedMonth = Table.AddColumn(AddedYear, "month", each Date.Month([InvoiceDate]), Int64.Type), // Flag returns AddedIsReturn = Table.AddColumn(AddedMonth, "is_return", each if Text.StartsWith([InvoiceNo], "C") then true else false, type logical), // Change data types ChangedTypes = Table.TransformColumnTypes(AddedIsReturn, { {"Quantity", Int64.Type}, {"UnitPrice", Currency.Type}, {"CustomerID", type text} }) in ChangedTypes ``` 4. **Data destination**: Lakehouse → `silver/online_retail_cleaned` 5. **Publish** the dataflow ### Refresh Dataflow Programmatically ```python # Using Fabric REST API import requests import os workspace_id = os.getenv("FABRIC_WORKSPACE_ID") dataflow_id = os.getenv("DATAFLOW_ID") access_token = os.getenv("FABRIC_ACCESS_TOKEN") url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/dataflows/{dataflow_id}/refresh" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } response = requests.post(url, headers=headers) print(f"Dataflow refresh status: {response.status_code}") ``` ## Fabric Notebooks: Silver to Gold with PySpark ### Revenue Trends Analysis ```python # Fabric Notebook - Gold Layer: Revenue Trends from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum, round, year, month, count spark = SparkSession.builder.getOrCreate() # Read Silver layer silver_df = spark.read.format("delta").load("Files/silver/online_retail_cleaned") # Filter out returns valid_sales = silver_df.filter(col("is_return") == False) # Aggregate revenue by year and month revenue_trends = valid_sales.groupBy("year", "month") \ .agg( sum("line_total").alias("total_revenue"), count("InvoiceNo").alias("total_transactions"), round(sum("line_total") / count("InvoiceNo"), 2).alias("avg_order_value") ) \ .orderBy("year", "month") # Write to Gold layer revenue_trends.write.format("delta") \ .mode("overwrite") \ .option("overwriteSchema", "true") \ .save("Files/gold/revenue_trends") # Display results display(revenue_trends) ``` ### Product Performance Analysis ```python # Fabric Notebook - Gold Layer: Product Performance from pyspark.sql.functions import col, sum, count, round, desc # Top products by revenue product_performance = valid_sales.groupBy("StockCode", "Description") \ .agg( sum("line_total").alias("total_revenue"), sum("Quantity").alias("total_quantity_sold"), count("InvoiceNo").alias("transaction_count"), round(sum("line_total") / sum("Quantity"), 2).alias("avg_unit_revenue") ) \ .filter(col("total_revenue") > 0) \ .orderBy(desc("total_revenue")) \ .limit(100) # Write to Gold layer product_performance.write.format("delta") \ .mode("overwrite") \ .save("Files/gold/product_performance") display(product_performance) ``` ### Customer Analytics ```python # Fabric Notebook - Gold Layer: Customer Analytics from pyspark.sql.functions import col, sum, count, countDistinct, round customer_analytics = valid_sales.groupBy("CustomerID") \ .agg( sum("line_total").alias("total_spent"), count("InvoiceNo").alias("total_orders"), countDistinct("StockCode").alias("unique_products_purchased"), round(sum("line_total") / count("InvoiceNo"), 2).alias("avg_order_value") ) \ .filter(col("total_spent") > 0) # Write to Gold layer customer_analytics.write.format("delta") \ .mode("overwrite") \ .save("Files/gold/customer_analytics") display(customer_analytics) ``` ### RFM Segmentation ```python # Fabric Notebook - Gold Layer: RFM Segmentation from pyspark.sql.functions import col, datediff, max, count, sum, lit, current_date from pyspark.sql.window import Window # Calculate RFM metrics max_date = valid_sales.select(max("InvoiceDate")).collect()[0][0] rfm = valid_sales.groupBy("CustomerID") \ .agg( datediff(lit(max_date), max("InvoiceDate")).alias("recency"), count("InvoiceNo").alias("frequency"), sum("line_total").alias("monetary") ) # Create RFM scores using ntile window_spec = Window.orderBy(col("recency").desc()) rfm_scored = rfm.withColumn("r_score", ntile(5).over(window_spec)) window_spec = Window.orderBy(col("frequency")) rfm_scored = rfm_scored.withColumn("f_score", ntile(5).over(window_spec)) window_spec = Window.orderBy(col("monetary")) rfm_scored = rfm_scored.withColumn("m_score", ntile(5).over(window_spec)) # Create RFM segment rfm_final = rfm_scored.withColumn("rfm_score", col("r_score") * 100 + col("f_score") * 10 + col("m_score")) # Segment classification from pyspark.sql.functions import when rfm_final = rfm_final.withColumn("customer_segment", when(col("rfm_score") >= 444, "Champions") .when(col("rfm_score") >= 334, "Loyal Customers") .when(col("rfm_score") >= 224, "Potential Loyalists") .when(col("rfm_score") >= 144, "At Risk") .otherwise("Lost") ) # Write to Gold layer rfm_final.write.format("delta") \ .mode("overwrite") \ .save("Files/gold/rfm_segmentation") display(rfm_final) ``` ## Creating Semantic Model ### 1. Create Semantic Model from Lakehouse 1. In Lakehouse explorer, go to **Tables** tab 2. Select Gold layer tables 3. Click **New semantic model** 4. Name: `RetailAnalyticsModel` ### 2. Define Relationships (DAX) Open the semantic model in Power BI Desktop or Fabric Model View: ```dax // Create Calendar table Calendar = ADDCOLUMNS( CALENDAR(DATE(2020, 1, 1), DATE(2023, 12, 31)), "Year", YEAR([Date]), "Month", MONTH([Date]), "MonthName", FORMAT([Date], "MMMM"), "Quarter", "Q" & ROUNDUP(MONTH([Date])/3, 0) ) // Create relationship: revenue_trends[year, month] → Calendar[Year, Month] ``` ### 3. Create Measures (DAX) ```dax // Total Revenue Total Revenue = SUM(revenue_trends[total_revenue]) // Total Transactions Total Transactions = SUM(revenue_trends[total_transactions]) // Average Order Value Avg Order Value = AVERAGE(revenue_trends[avg_order_value]) // Revenue Growth % Revenue Growth % = VAR CurrentRevenue = [Total Revenue] VAR PreviousRevenue = CALCULATE( [Total Revenue], DATEADD(Calendar[Date], -1, MONTH) ) RETURN DIVIDE(CurrentRevenue - PreviousRevenue, PreviousRevenue, 0) // Customer Lifetime Value Customer LTV = AVERAGE(customer_analytics[total_spent]) // Top Product Revenue Top Product Revenue = CALCULATE( SUM(product_performance[total_revenue]), TOPN(1, ALL(product_performance), product_performance[total_revenue]) ) ``` ## Creating Power BI Reports ### Connect to Semantic Model ```python # Power BI connects directly to the Fabric semantic model # No additional code needed - use the Power BI service or Desktop ```