Building a Production-Ready Data Pipeline with Azure (Part 9): Migrating from Synapse Serverless SQL

# Fabric notebook for automated shortcut creation
# This runs inside a Fabric notebook with authentication already configured

print("="*60) print("FABRIC MIGRATION: AUTOMATED SHORTCUT CREATION") print("="*60) import pandas as pd
import sempy.fabric as fabric
from notebookutils import mssparkutils
import requests
import time
from datetime import datetime
# Fabric authentication
url = 'https://api.fabric.microsoft.com/'
access_token = mssparkutils.credentials.getToken(url)
# Configuration - Replace with your values
workspaceId = 'your-workspace-id'
lakehouseid = 'your-lakehouse-id'  
targetconnectionid = 'your-connection-id'  
targetlocation = 'https://yourstorage.dfs.core.windows.net'
headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json"
}
print("Creating API shortcuts under dbo schema with drop-if-exists logic...") api_url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{lakehouseid}/shortcuts"
# Define your star schema tables
tables_to_create = [
    ("dim_customer", "gold/star_schema/dim_customer"),
    ("dim_date", "gold/star_schema/dim_date"),
    ("dim_product", "gold/star_schema/dim_product"),
    ("dim_territory", "gold/star_schema/dim_territory"),
    ("dim_salesperson", "gold/star_schema/dim_salesperson"),
    ("dim_store", "gold/star_schema/dim_store"),
    ("dim_promotion", "gold/star_schema/dim_promotion"),
    ("fact_sales", "gold/star_schema/fact_sales"),
    ("fact_sales_summary", "gold/star_schema/fact_sales_summary")
]
successful_shortcuts = []
failed_shortcuts = []
for shortcutname, goldpath in tables_to_create:
    print(f"\nProcessing: dbo.{shortcutname}")
    
    # Step 1: Check if shortcut exists
    check_url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{lakehouseid}/shortcuts/Tables/dbo/{shortcutname}"
    check_response = requests.get(check_url, headers=headers)
    
    # Step 2: If exists, delete it first
    if check_response.status_code == 200:
        print(f"   Shortcut exists, deleting first...") delete_response = requests.delete(check_url, headers=headers) if delete_response.status_code == 200:
            print(f"   Successfully deleted existing shortcut") time.sleep(2)  # Wait for deletion to complete
        else:
            print(f"   Failed to delete: {delete_response.status_code} - {delete_response.text}")
    
    # Step 3: Create new shortcut
    payload = {
        "path": "Tables/dbo",  # Create under dbo schema
        "name": shortcutname,
        "target": {
            "AdlsGen2": {
                "location": targetlocation,
                "subpath": f"datalake/{goldpath}",
                "connectionId": targetconnectionid
            }
        }
    }
    
    response = requests.post(api_url, headers=headers, json=payload) if response.status_code == 201:
        print(f"   SUCCESS: dbo.{shortcutname} created") successful_shortcuts.append(shortcutname) elif response.status_code == 409:
        print(f"   ALREADY EXISTS: dbo.{shortcutname}") successful_shortcuts.append(shortcutname) else:
        print(f"   FAILED: dbo.{shortcutname} - {response.status_code}") print(f"   Error: {response.text}") failed_shortcuts.append(shortcutname) time.sleep(2)
# Summary
print(f"\n" + "="*60) print("SHORTCUT CREATION SUMMARY") print("="*60) print(f"Successful: {len(successful_shortcuts)}") print(f"Failed: {len(failed_shortcuts)}") if failed_shortcuts:
    print(f"\nFailed shortcuts:") for shortcut in failed_shortcuts:
        print(f"  - {shortcut}") print("\nWaiting for shortcuts to sync...") time.sleep(15)
# Test the shortcuts
print("\nTesting dbo schema shortcuts...") for shortcutname, _ in tables_to_create:
    try:
        # Test access via dbo schema path
        df = spark.read.format("delta").load(f"Tables/dbo/{shortcutname}") count = df.count() print(f"   dbo.{shortcutname}: {count:,} rows - SUCCESS")
        
        # Create temp view for SQL access
        df.createOrReplaceTempView(shortcutname) except Exception as e:
        print(f"   dbo.{shortcutname}: Failed - {str(e)}") print("\nTesting SQL queries...") try:
    result = spark.sql("SELECT COUNT(*) as cnt FROM dim_customer").collect()[0].cnt
    print(f"SQL test: dim_customer has {result:,} rows") except Exception as e:
    print(f"SQL test failed: {str(e)}")

// Update these parameters in Power Query
ServerName = "your-fabric-sql-endpoint.datawarehouse.fabric.microsoft.com"
DatabaseName = "your_lakehouse_name"
SchemaName = "dbo"

# Validation script to compare data between Synapse and Fabric
def validate_migration():
    """
    Compare row counts and sample data between 
    Synapse Serverless SQL Pool and Fabric
    """
    
    validation_results = []
    
    for table_name, _ in tables_to_create:
        try:
            # Get row count from Fabric
            fabric_count = spark.sql(f"SELECT COUNT(*) as cnt FROM {table_name}").collect()[0].cnt
            
            # Get sample data for validation
            sample_df = spark.sql(f"SELECT * FROM {table_name} LIMIT 5") validation_results.append({
                'table': table_name,
                'row_count': fabric_count,
                'status': 'SUCCESS',
                'sample_rows': sample_df.count()
            }) except Exception as e:
            validation_results.append({
                'table': table_name,
                'row_count': 0,
                'status': 'FAILED',
                'error': str(e)
            })
    
    # Create validation report
    validation_df = pd.DataFrame(validation_results) print("\nVALIDATION REPORT") print("="*60) print(validation_df.to_string()) return validation_df

# Run validation
validation_report = validate_migration()

Synapse Serverless SQL Pool:
- Data Processed: 10 TB/month
- Cost: $5/TB = $50
- Additional: Synapse workspace overhead

Microsoft Fabric:
- Capacity: F2 SKU
- Cost: ~$0.18/hour = ~$130/month
- Includes: Compute, Storage, Power BI

Result: Similar cost but with significantly better performance and features

# Apply V-Order optimization to Delta tables
from delta.tables import DeltaTable
for table_name, delta_path in tables_to_create:
    table_path = f"abfss://datalake@yourstorage.dfs.core.windows.net/{delta_path}"
    
    # Optimize with Z-Order
    delta_table = DeltaTable.forPath(spark, table_path) delta_table.optimize().executeZOrderBy("date_key", "customer_key") print(f"Optimized: {table_name}")