❄️ Tutorial 24: Snowflake to Microsoft Fabric Migration¶

# PySpark: Read VARIANT data as JSON strings
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("machine_id", StringType(), False),
    StructField("event_timestamp", TimestampType(), False),
    StructField("event_payload", StringType(), True),    # JSON string
    StructField("player_tags", StringType(), True),      # JSON array string
    StructField("machine_config", StringType(), True),   # JSON object string
])

# Query JSON fields using Spark SQL after loading
df = spark.table("bronze.slot_events")
df_parsed = df.selectExpr(
    "event_id",
    "machine_id",
    "event_timestamp",
    "get_json_object(event_payload, '$.spin_result') AS spin_result",
    "get_json_object(event_payload, '$.bet_amount') AS bet_amount",
    "get_json_object(event_payload, '$.payout_amount') AS payout_amount"
)

# Fabric Notebook: Test Snowflake Connectivity
from pyspark.sql import SparkSession
from notebookutils import mssparkutils

# Retrieve credentials from Azure Key Vault
sf_account = "xy12345.us-east-1"
sf_user = mssparkutils.credentials.getSecret("casino-keyvault", "snowflake-user")
sf_password = mssparkutils.credentials.getSecret("casino-keyvault", "snowflake-password")

# Test JDBC connection
jdbc_url = f"jdbc:snowflake://{sf_account}.snowflakecomputing.com/"

try:
    df_test = spark.read \
        .format("jdbc") \
        .option("url", jdbc_url) \
        .option("user", sf_user) \
        .option("password", sf_password) \
        .option("db", "CASINO_DW") \
        .option("schema", "PUBLIC") \
        .option("warehouse", "CASINO_ETL_WH") \
        .option("query", "SELECT CURRENT_TIMESTAMP() AS connected_at, CURRENT_DATABASE() AS db, CURRENT_WAREHOUSE() AS wh") \
        .option("driver", "net.snowflake.client.jdbc.SnowflakeDriver") \
        .load()

    df_test.show()
    print("Snowflake connection successful!")
except Exception as e:
    print(f"Connection failed: {e}")

from pyspark.sql.functions import explode, from_json, col
from pyspark.sql.types import ArrayType, StringType

df = spark.table("casino.player_profiles")

# Parse JSON array string and explode
df_tags = df.withColumn(
    "loyalty_tags_parsed",
    from_json(col("loyalty_tags"), ArrayType(StringType()))
).select(
    "player_id",
    "player_name",
    explode("loyalty_tags_parsed").alias("loyalty_tag")
).filter(col("loyalty_tag").like("%VIP%"))

# Fabric Notebook: refresh_loyalty_tiers (scheduled via pipeline)
# Schedule: Daily at 6:00 AM PT via Data Factory pipeline trigger

from pyspark.sql.functions import col, sum as spark_sum, when, current_timestamp
from delta.tables import DeltaTable

# Read transaction history
df_transactions = spark.table("silver.player_transactions")

# Compute loyalty tiers
df_tiers = df_transactions.groupBy("player_id").agg(
    spark_sum("total_spend").alias("lifetime_spend")
).withColumn("computed_tier",
    when(col("lifetime_spend") >= 100000, "PLATINUM")
    .when(col("lifetime_spend") >= 50000, "GOLD")
    .when(col("lifetime_spend") >= 10000, "SILVER")
    .otherwise("BRONZE")
).withColumn("updated_at", current_timestamp())

# MERGE into target (Delta Lake MERGE)
target = DeltaTable.forName(spark, "gold.player_loyalty_tiers")

target.alias("t").merge(
    df_tiers.alias("s"),
    "t.player_id = s.player_id"
).whenMatchedUpdate(
    condition="t.tier != s.computed_tier",
    set={"tier": "s.computed_tier", "updated_at": "s.updated_at"}
).whenNotMatchedInsert(
    values={
        "player_id": "s.player_id",
        "tier": "s.computed_tier",
        "updated_at": "s.updated_at"
    }
).execute()

print("Loyalty tiers refreshed successfully")

# Fabric Notebook: calculate_ctr_report
# Parameters cell (for pipeline integration)
report_date = "2024-06-15"  # Overridden by pipeline parameter

from pyspark.sql.functions import col, sum as spark_sum, when, lit

# CTR threshold: $10,000 per NIGC MICS compliance
CTR_THRESHOLD = 10000

df_cage = spark.table("silver.cage_transactions") \
    .filter(col("transaction_date") == report_date)

df_ctr = df_cage.groupBy("transaction_date", "player_id").agg(
    spark_sum("amount").alias("total_amount")
).filter(col("total_amount") >= CTR_THRESHOLD) \
 .withColumn("ctr_status", lit("CTR_REQUIRED"))

# Append to CTR reports table
df_ctr.write.mode("append").saveAsTable("gold.ctr_reports")

print(f"CTR report generated for {report_date}: {df_ctr.count()} records")

# Enable Change Data Feed on the source table
spark.sql("""
    ALTER TABLE bronze.player_profiles
    SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# Read changes since last checkpoint (like Snowflake STREAM consumption)
df_changes = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 5) \
    .table("bronze.player_profiles")

# Filter to inserts and updates only
df_upserts = df_changes.filter(
    col("_change_type").isin("insert", "update_postimage")
)

# MERGE into silver (like Snowflake TASK consuming a STREAM)
from delta.tables import DeltaTable
target = DeltaTable.forName(spark, "silver.player_profiles")
target.alias("t").merge(
    df_upserts.alias("s"),
    "t.player_id = s.player_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Fabric Notebook: Incremental Snowflake Migration
from pyspark.sql.functions import max as spark_max, col
from datetime import datetime

# Configuration
sf_table = "CASINO_DW.PLAYER_TRANSACTIONS"
fabric_table = "bronze.player_transactions"
watermark_column = "transaction_timestamp"

# Get current watermark from Fabric
try:
    current_watermark = spark.table(fabric_table) \
        .select(spark_max(watermark_column)) \
        .collect()[0][0]
    watermark_str = current_watermark.strftime('%Y-%m-%d %H:%M:%S')
except Exception:
    watermark_str = "2020-01-01 00:00:00"

print(f"Current watermark: {watermark_str}")

# Read incremental data from Snowflake using CHANGES clause
query = f"""
    SELECT *
    FROM {sf_table}
        CHANGES (INFORMATION => DEFAULT)
        AT (TIMESTAMP => '{watermark_str}'::TIMESTAMP)
"""

df_incremental = spark.read \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("query", query) \
    .option("user", sf_user) \
    .option("password", sf_password) \
    .option("driver", "net.snowflake.client.jdbc.SnowflakeDriver") \
    .load()

# Append to Fabric table
row_count = df_incremental.count()
print(f"Rows to migrate: {row_count:,}")

if row_count > 0:
    df_incremental.write.mode("append").saveAsTable(fabric_table)
    print(f"Successfully migrated {row_count:,} rows")

# Fabric Notebook: Partitioned Large Table Migration
from pyspark.sql.functions import col, year, month

# Configuration
sf_table = "CASINO_DW.SLOT_TRANSACTIONS_HISTORY"
fabric_table = "bronze.slot_transactions_history"

# Partitioned JDBC read for parallel extraction
df = spark.read \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", sf_table) \
    .option("user", sf_user) \
    .option("password", sf_password) \
    .option("driver", "net.snowflake.client.jdbc.SnowflakeDriver") \
    .option("partitionColumn", "TRANSACTION_ID") \
    .option("lowerBound", "1") \
    .option("upperBound", "500000000") \
    .option("numPartitions", 48) \
    .option("fetchsize", 100000) \
    .load()

print(f"Extracting with 48 parallel partitions...")

# Write with Fabric partitioning
df.write \
    .mode("overwrite") \
    .partitionBy("transaction_year", "transaction_month") \
    .format("delta") \
    .saveAsTable(fabric_table)

# Optimize the table
spark.sql(f"OPTIMIZE {fabric_table} ZORDER BY (machine_id, player_id)")
print("Migration and optimization complete")

# Fabric Notebook: Load staged Parquet files from ADLS
# (Assumes Snowflake external stage points to an Azure storage account)

storage_path = "abfss://migration@casinostorage.dfs.core.windows.net/slot_transactions/"

df = spark.read.parquet(storage_path)

print(f"Loaded {df.count():,} rows from staged Parquet files")

# Write to Lakehouse
df.write.mode("overwrite").format("delta").saveAsTable("bronze.slot_transactions")

# Clean up staging area after validation
# mssparkutils.fs.rm(storage_path, recurse=True)

# Fabric Notebook: Configure Eventstream consumer
# This replaces Snowflake's CREATE STREAM + CREATE TASK pattern

# Eventstream automatically ingests from Event Hub / Kafka
# Configure in Fabric portal: Eventstream > Add source > Azure Event Hub

# Read from Eventstream-backed KQL database for real-time analytics
df_realtime = spark.sql("""
    SELECT
        machine_id,
        player_id,
        event_type,
        event_timestamp,
        coin_in,
        coin_out
    FROM eventhouse.slot_telemetry
    WHERE event_timestamp > DATEADD(MINUTE, -5, CURRENT_TIMESTAMP())
""")

display(df_realtime)

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def mask_ssn(ssn):
    """Mask SSN for compliance display -- shows only last 4 digits."""
    if ssn is None or len(ssn) < 4:
        return "***-**-****"
    return f"***-**-{ssn[-4:]}"

# Register for SQL usage
spark.udf.register("mask_ssn", mask_ssn)

# DataFrame API usage
df = spark.table("bronze.player_pii")
df_masked = df.withColumn("masked_ssn", mask_ssn("ssn"))

# Spark SQL usage
spark.sql("SELECT player_id, mask_ssn(ssn) AS masked_ssn FROM bronze.player_pii")

from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType

@udf(returnType=DoubleType())
def calculate_loyalty_points(total_spend, tier, multiplier_event):
    """Calculate loyalty points based on tier and event multipliers."""
    tier_rates = {
        "PLATINUM": 3.0,
        "GOLD": 2.0,
        "SILVER": 1.5,
        "BRONZE": 1.0
    }
    base_rate = tier_rates.get(tier, 1.0)
    points = (total_spend or 0.0) * base_rate
    if multiplier_event:
        points *= 2
    return round(points, 2)

spark.udf.register("calculate_loyalty_points", calculate_loyalty_points)

# Apply to player transactions
df = spark.table("silver.player_transactions")
df_points = df.withColumn(
    "loyalty_points",
    calculate_loyalty_points("total_spend", "tier", "multiplier_event")
)

# Register as Spark SQL function
spark.sql("""
    CREATE OR REPLACE FUNCTION casino.win_loss_ratio(coin_in DECIMAL(18,2), coin_out DECIMAL(18,2))
    RETURNS DECIMAL(10,4)
    RETURN CASE WHEN coin_in = 0 THEN 0 ELSE coin_out / coin_in END
""")

import requests
from pyspark.sql.functions import udf, struct, to_json
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def check_fraud(transaction_json):
    """Call fraud detection API -- equivalent to Snowflake external function."""
    try:
        response = requests.post(
            "https://fraud-api.casino.com/v1/check",
            json={"transaction": transaction_json},
            headers={"Authorization": f"Bearer {api_token}"},
            timeout=5
        )
        return response.text
    except Exception as e:
        return f'{{"error": "{str(e)}"}}'

# Apply to suspicious transactions
df_suspicious = spark.table("silver.cage_transactions") \
    .filter("amount >= 8000 AND amount <= 9999")  # SAR structuring threshold

df_checked = df_suspicious.withColumn(
    "fraud_result",
    check_fraud(to_json(struct("*")))
)

# Fabric Notebook: Row Count Validation
from pyspark.sql import SparkSession
import pandas as pd

# Snowflake JDBC settings
sf_jdbc_url = f"jdbc:snowflake://{sf_account}.snowflakecomputing.com/"
sf_properties = {
    "user": sf_user,
    "password": sf_password,
    "db": "CASINO_DW",
    "schema": "PUBLIC",
    "warehouse": "CASINO_ETL_WH",
    "driver": "net.snowflake.client.jdbc.SnowflakeDriver"
}

# Tables to validate
validation_tables = [
    ("CASINO_DW.SLOT_TRANSACTIONS", "bronze.slot_transactions"),
    ("CASINO_DW.PLAYER_PROFILES", "bronze.player_profiles"),
    ("CASINO_DW.CAGE_OPERATIONS", "bronze.cage_operations"),
    ("CASINO_DW.LOYALTY_EVENTS", "bronze.loyalty_events"),
    ("CASINO_DW.TABLE_GAME_SESSIONS", "bronze.table_game_sessions"),
]

results = []

for sf_table, fabric_table in validation_tables:
    # Fabric count
    fabric_count = spark.table(fabric_table).count()

    # Snowflake count
    sf_count_df = spark.read \
        .format("jdbc") \
        .option("url", sf_jdbc_url) \
        .options(**sf_properties) \
        .option("query", f"SELECT COUNT(*) AS cnt FROM {sf_table}") \
        .load()
    sf_count = sf_count_df.collect()[0]["cnt"]

    # Calculate difference
    diff = abs(sf_count - fabric_count)
    diff_pct = (diff / sf_count * 100) if sf_count > 0 else 0
    status = "PASS" if diff_pct < 0.01 else "FAIL"

    results.append({
        "Snowflake Table": sf_table,
        "Fabric Table": fabric_table,
        "Snowflake Count": f"{sf_count:,}",
        "Fabric Count": f"{fabric_count:,}",
        "Difference": f"{diff:,}",
        "Diff %": f"{diff_pct:.4f}%",
        "Status": status
    })

df_results = pd.DataFrame(results)
display(df_results)

# Fabric Notebook: Numeric Checksum Validation
from pyspark.sql.functions import sum as spark_sum, count, avg, min as spark_min, max as spark_max

def calculate_checksums(df, numeric_columns):
    """Calculate aggregate checksums for numeric columns."""
    agg_exprs = [count("*").alias("row_count")]
    for col_name in numeric_columns:
        agg_exprs.extend([
            spark_sum(col_name).alias(f"sum_{col_name}"),
            avg(col_name).alias(f"avg_{col_name}"),
            spark_min(col_name).alias(f"min_{col_name}"),
            spark_max(col_name).alias(f"max_{col_name}")
        ])
    return df.agg(*agg_exprs).collect()[0]

# Validate slot_transactions
df_fabric = spark.table("bronze.slot_transactions")
numeric_cols = ["coin_in", "coin_out", "jackpot_contribution", "theoretical_win"]

fabric_checksums = calculate_checksums(df_fabric, numeric_cols)

print("=== Fabric Checksums ===")
print(f"  Row Count:       {fabric_checksums['row_count']:,}")
for col_name in numeric_cols:
    print(f"  Sum {col_name:25s}: {fabric_checksums[f'sum_{col_name}']:,.2f}")
    print(f"  Avg {col_name:25s}: {fabric_checksums[f'avg_{col_name}']:,.4f}")

# Run equivalent checksums on Snowflake for comparison
sf_checksum_query = f"""
    SELECT
        COUNT(*) AS row_count,
        {', '.join([f"SUM({c}) AS sum_{c}, AVG({c}) AS avg_{c}" for c in numeric_cols])}
    FROM CASINO_DW.SLOT_TRANSACTIONS
"""

df_sf_checksums = spark.read.format("jdbc") \
    .option("url", sf_jdbc_url) \
    .options(**sf_properties) \
    .option("query", sf_checksum_query) \
    .load()

print("\n=== Snowflake Checksums ===")
display(df_sf_checksums)

# Fabric Notebook: Schema Comparison
from pyspark.sql.functions import col
import pandas as pd

def compare_schemas(sf_table, fabric_table):
    """Compare column names and types between Snowflake and Fabric tables."""
    # Get Snowflake schema
    sf_schema_query = f"""
        SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, CHARACTER_MAXIMUM_LENGTH, NUMERIC_PRECISION, NUMERIC_SCALE
        FROM INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_SCHEMA || '.' || TABLE_NAME = '{sf_table}'
        ORDER BY ORDINAL_POSITION
    """
    df_sf_schema = spark.read.format("jdbc") \
        .option("url", sf_jdbc_url) \
        .options(**sf_properties) \
        .option("query", sf_schema_query) \
        .load().toPandas()

    # Get Fabric schema
    df_fabric = spark.table(fabric_table)
    fabric_cols = [(f.name, str(f.dataType), f.nullable) for f in df_fabric.schema.fields]
    df_fabric_schema = pd.DataFrame(fabric_cols, columns=["COLUMN_NAME", "DATA_TYPE", "NULLABLE"])

    # Compare
    print(f"\n=== Schema Comparison: {sf_table} -> {fabric_table} ===")
    print(f"Snowflake columns: {len(df_sf_schema)}")
    print(f"Fabric columns:    {len(df_fabric_schema)}")

    sf_cols = set(df_sf_schema["COLUMN_NAME"].str.lower())
    fb_cols = set(df_fabric_schema["COLUMN_NAME"].str.lower())

    missing_in_fabric = sf_cols - fb_cols
    extra_in_fabric = fb_cols - sf_cols

    if missing_in_fabric:
        print(f"  Missing in Fabric: {missing_in_fabric}")
    if extra_in_fabric:
        print(f"  Extra in Fabric:   {extra_in_fabric}")
    if not missing_in_fabric and not extra_in_fabric:
        print("  All columns present in both systems")

    return df_sf_schema, df_fabric_schema

# Run for all migrated tables
for sf_table, fabric_table in validation_tables:
    compare_schemas(sf_table, fabric_table)