🔄 Tutorial 10: Teradata to Microsoft Fabric Migration¶

# Fabric Notebook: Test Teradata connectivity via Data Factory
from notebookutils import mssparkutils

# Test connection using Data Factory linked service
connection_result = mssparkutils.data.connect(
    linkedService="ls_teradata_casino_dw"
)
print(f"Connection Status: {connection_result}")

from pyspark.sql.functions import row_number, desc
from pyspark.sql.window import Window

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

window_spec = Window.partitionBy("player_id").orderBy(desc("session_timestamp"))

df_latest = df.withColumn("rn", row_number().over(window_spec)) \
    .filter("rn = 1") \
    .drop("rn")

# PySpark: Remove duplicates during ingestion
df = spark.read.parquet("Files/raw/slot_events/")

# Remove exact duplicates
df_deduped = df.dropDuplicates()

# Or remove duplicates by key
df_deduped = df.dropDuplicates(["event_id"])

df_deduped.write.mode("overwrite").saveAsTable("casino.slot_events")

# Top 100 with ties
from pyspark.sql.functions import sum, rank
from pyspark.sql.window import Window

df = spark.table("casino.player_transactions")
df_agg = df.groupBy("player_id").agg(sum("total_spend").alias("lifetime_spend"))
window = Window.orderBy(desc("lifetime_spend"))
df_ranked = df_agg.withColumn("rnk", rank().over(window))
df_top = df_ranked.filter("rnk <= 100").drop("rnk")

# Random sample
df_sample = spark.table("casino.large_table").sample(0.01)

# Fabric Notebook: Incremental Teradata Migration
from pyspark.sql import SparkSession
from pyspark.sql.functions import max as spark_max
from datetime import datetime

# Configuration
teradata_table = "CASINO_DW.SLOT_TRANSACTIONS"
fabric_table = "bronze.slot_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]
except:
    current_watermark = datetime(2020, 1, 1)  # Default start

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

# Read incremental data from Teradata
query = f"""
SELECT * FROM {teradata_table}
WHERE {watermark_column} > TIMESTAMP '{current_watermark}'
ORDER BY {watermark_column}
"""

# Use JDBC connection (requires Teradata JDBC driver)
df_incremental = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:teradata://teradata.casino.com/CASINO_DW") \
    .option("dbtable", f"({query}) AS subq") \
    .option("user", mssparkutils.credentials.getSecret("keyvault", "teradata-user")) \
    .option("password", mssparkutils.credentials.getSecret("keyvault", "teradata-password")) \
    .option("driver", "com.teradata.jdbc.TeraDriver") \
    .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
teradata_table = "CASINO_DW.SLOT_TRANSACTIONS_HISTORY"
fabric_table = "bronze.slot_transactions_history"
partition_column = "transaction_date"

# Read with partition pruning
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:teradata://teradata.casino.com/CASINO_DW") \
    .option("dbtable", teradata_table) \
    .option("user", mssparkutils.credentials.getSecret("keyvault", "teradata-user")) \
    .option("password", mssparkutils.credentials.getSecret("keyvault", "teradata-password")) \
    .option("partitionColumn", partition_column) \
    .option("lowerBound", "2020-01-01") \
    .option("upperBound", "2024-12-31") \
    .option("numPartitions", 48) \
    .option("fetchsize", 100000) \
    .load()

# 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}")

# Cell 1: Configuration
# ====================
from datetime import datetime, timedelta
from pyspark.sql.functions import col, count, sum as spark_sum

# Parameters
report_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
export_path = f"Files/exports/daily_slots_{report_date}.csv"

print(f"Processing report for: {report_date}")

# Cell 2: Read Source Data
# ========================
# Option A: Read from already-migrated Lakehouse table
df_transactions = spark.table("bronze.slot_transactions")

# Option B: Read directly from Teradata (if still connected)
# df_transactions = spark.read.format("jdbc").option(...).load()

# Cell 3: Transform - Daily Slot Summary
# ======================================
df_daily_summary = df_transactions \
    .filter(col("transaction_date") == report_date) \
    .groupBy("machine_id") \
    .agg(
        count("*").alias("spin_count"),
        spark_sum("coin_in").alias("total_coin_in"),
        spark_sum("coin_out").alias("total_coin_out")
    ) \
    .withColumn("report_date", lit(report_date)) \
    .orderBy(col("total_coin_in").desc())

# Cell 4: Export to CSV
# =====================
df_daily_summary.coalesce(1) \
    .write \
    .mode("overwrite") \
    .option("header", "true") \
    .csv(export_path)

print(f"Exported {df_daily_summary.count()} rows to {export_path}")

# Cell 5: Save to Silver Layer
# ============================
df_daily_summary.write \
    .mode("append") \
    .saveAsTable("silver.daily_slot_summary")

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

# Tables to validate
validation_tables = [
    ("CASINO_DW.SLOT_TRANSACTIONS", "bronze.slot_transactions"),
    ("CASINO_DW.PLAYER_SESSIONS", "bronze.player_sessions"),
    ("CASINO_DW.CAGE_OPERATIONS", "bronze.cage_operations"),
]

results = []

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

    # Get Teradata count (via JDBC)
    teradata_count_df = spark.read \
        .format("jdbc") \
        .option("url", "jdbc:teradata://teradata.casino.com/CASINO_DW") \
        .option("query", f"SELECT COUNT(*) AS cnt FROM {teradata_table}") \
        .load()
    teradata_count = teradata_count_df.collect()[0]["cnt"]

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

    results.append({
        "Source Table": teradata_table,
        "Target Table": fabric_table,
        "Source Count": teradata_count,
        "Target Count": fabric_count,
        "Difference": diff,
        "Diff %": f"{diff_pct:.4f}%",
        "Status": status
    })

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

# Fabric Notebook: Checksum Validation
from pyspark.sql.functions import sum as spark_sum, count, md5, concat_ws

def calculate_table_checksum(df, key_columns, numeric_columns):
    """Calculate a checksum for data validation."""
    return df.agg(
        count("*").alias("row_count"),
        *[spark_sum(col).alias(f"sum_{col}") for col in numeric_columns]
    ).collect()[0]

# Compare slot_transactions
df_fabric = spark.table("bronze.slot_transactions")
fabric_checksum = calculate_table_checksum(
    df_fabric,
    key_columns=["transaction_id"],
    numeric_columns=["coin_in", "coin_out", "jackpot_contribution"]
)

print("Fabric Checksum:")
print(f"  Row Count: {fabric_checksum['row_count']:,}")
print(f"  Sum Coin In: ${fabric_checksum['sum_coin_in']:,.2f}")
print(f"  Sum Coin Out: ${fabric_checksum['sum_coin_out']:,.2f}")

# Compare against Teradata (run equivalent query on source)

# Fabric Notebook: Sample Row Comparison
from pyspark.sql.functions import col

# Get sample records from Fabric
df_fabric_sample = spark.table("bronze.slot_transactions") \
    .filter(col("transaction_date") == "2024-01-15") \
    .limit(100) \
    .toPandas()

# Get same sample from Teradata
query = """
SELECT * FROM CASINO_DW.SLOT_TRANSACTIONS
WHERE transaction_date = DATE '2024-01-15'
SAMPLE 100
"""
df_teradata_sample = spark.read.format("jdbc") \
    .option("query", query) \
    .load() \
    .toPandas()

# Compare schemas
print("Schema Comparison:")
print(f"Fabric columns: {list(df_fabric_sample.columns)}")
print(f"Teradata columns: {list(df_teradata_sample.columns)}")

# Merge and compare (on key column)
comparison = df_fabric_sample.merge(
    df_teradata_sample,
    on="transaction_id",
    how="outer",
    suffixes=("_fabric", "_teradata"),
    indicator=True
)

print(f"\nMatching records: {len(comparison[comparison['_merge'] == 'both'])}")
print(f"Fabric only: {len(comparison[comparison['_merge'] == 'left_only'])}")
print(f"Teradata only: {len(comparison[comparison['_merge'] == 'right_only'])}")

# Cell 1: Configuration
# ====================
from pyspark.sql.functions import col, current_timestamp
from delta.tables import DeltaTable

staging_path = "Files/feeds/loyalty_daily_update.csv"
target_table = "silver.player_loyalty_accounts"

# Cell 2: Read staging data (replaces .IMPORT INFILE)
# ====================================================
df_updates = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv(staging_path) \
    .withColumn("_loaded_at", current_timestamp())

print(f"Staging records to merge: {df_updates.count():,}")

# Cell 3: Delta Lake MERGE (replaces MultiLoad DML)
# ==================================================
target = DeltaTable.forName(spark, target_table)

target.alias("tgt").merge(
    df_updates.alias("src"),
    "tgt.player_id = src.player_id"
).whenMatchedUpdate(set={
    "loyalty_points": "src.loyalty_points",
    "tier_status":    "src.tier_status",
    "last_visit":     "src.last_visit",
    "_loaded_at":     "src._loaded_at"
}).whenNotMatchedInsert(values={
    "player_id":      "src.player_id",
    "loyalty_points": "src.loyalty_points",
    "tier_status":    "src.tier_status",
    "last_visit":     "src.last_visit",
    "_loaded_at":     "src._loaded_at"
}).execute()

# Cell 4: Post-merge validation (replaces .LOGTABLE check)
# =========================================================
history = spark.sql(f"DESCRIBE HISTORY {target_table} LIMIT 1").collect()[0]
print(f"Operation: {history['operation']}")
print(f"Rows updated: {history['operationMetrics']['numTargetRowsUpdated']}")
print(f"Rows inserted: {history['operationMetrics']['numTargetRowsInserted']}")

# Cell 1: Configuration
# ====================
from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import (
    StructType, StructField, StringType, TimestampType, DecimalType
)

# Eventstream connection (replaces TPump INFILE)
eh_connection_string = mssparkutils.credentials.getSecret(
    "keyvault", "slot-events-eventhub-conn"
)
target_table = "bronze.slot_events_rt"
checkpoint_path = "Files/_checkpoints/slot_events_rt"

# Cell 2: Define schema (replaces .LAYOUT)
# =========================================
slot_event_schema = StructType([
    StructField("machine_id", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("event_ts", TimestampType(), False),
    StructField("coin_in", DecimalType(12, 2), True),
    StructField("coin_out", DecimalType(12, 2), True)
])

# Cell 3: Structured Streaming (replaces TPump continuous load)
# =============================================================
df_stream = spark.readStream \
    .format("eventhubs") \
    .option("eventhubs.connectionString",
            sc._jvm.org.apache.spark.eventhubs.EventHubsUtils
            .encrypt(eh_connection_string)) \
    .option("eventhubs.startingPosition",
            '{"offset":"-1","seqNo":-1,"enqueuedTime":null,"isInclusive":true}') \
    .load()

# Parse JSON payload into typed columns
df_parsed = df_stream \
    .select(from_json(col("body").cast("string"), slot_event_schema).alias("data")) \
    .select("data.*") \
    .withColumn("_ingest_ts", current_timestamp())

# Cell 4: Write to Delta Lake (replaces INSERT DML)
# ==================================================
# Micro-batch append mirrors TPump SLEEP/PACK cadence
query = df_parsed.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", checkpoint_path) \
    .trigger(processingTime="60 seconds") \
    .toTable(target_table)

print(f"Streaming query started: {query.id}")
print(f"Processing every 60 seconds (equivalent to TPump SLEEP 60)")

# Cell 1: Configuration
# ====================
from pyspark.sql.functions import col, lit, current_timestamp, when
from datetime import datetime, timedelta
import sys

report_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
staging_table = "bronze.stg_cage_transactions"
target_table = "silver.cage_transactions_daily"
ctr_threshold = 10000

exit_code = 0  # Mirrors .QUIT return codes

# Cell 2: Step 1 - Load staging (replaces DELETE + INSERT + .IF ACTIVITYCOUNT)
# =============================================================================
try:
    # Clear staging (replaces DELETE FROM)
    spark.sql(f"TRUNCATE TABLE {staging_table}")

    # Load from raw
    df_raw = spark.table("bronze.cage_transactions_raw") \
        .filter(col("transaction_date") == report_date)

    staging_count = df_raw.count()

    if staging_count == 0:
        print(f"WARNING: No data found for {report_date}")
        exit_code = 4  # Mirrors .QUIT 4 (NO_DATA_EXIT)
        mssparkutils.notebook.exit(f"NO_DATA|exit_code={exit_code}")

    df_raw.write.mode("overwrite").saveAsTable(staging_table)
    print(f"Step 1 PASS: {staging_count:,} rows loaded to staging")

except Exception as e:
    print(f"Step 1 FAIL: {str(e)}")
    exit_code = 8  # Mirrors .QUIT 8 (ERROR_EXIT)
    raise

# Cell 3: Step 2 - CTR flagging (replaces UPDATE + .IF ERRORCODE)
# ================================================================
try:
    spark.sql(f"""
        UPDATE {staging_table}
        SET ctr_flag = 'Y'
        WHERE transaction_amount >= {ctr_threshold}
    """)

    ctr_count = spark.table(staging_table) \
        .filter(col("ctr_flag") == "Y").count()
    print(f"Step 2 PASS: {ctr_count:,} transactions flagged for CTR")

except Exception as e:
    print(f"Step 2 FAIL: {str(e)}")
    exit_code = 8
    raise

# Cell 4: Step 3 - Move to production (replaces INSERT + .IF ACTIVITYCOUNT check)
# =================================================================================
try:
    df_staging = spark.table(staging_table)
    pre_count = df_staging.count()

    df_staging.write.mode("append").saveAsTable(target_table)

    # Validate row count (replaces .IF ACTIVITYCOUNT <> SELECT COUNT)
    post_count = spark.table(target_table) \
        .filter(col("transaction_date") == report_date).count()

    if post_count != pre_count:
        print(f"Step 3 FAIL: Count mismatch - staging={pre_count:,}, target={post_count:,}")
        exit_code = 12  # Mirrors .QUIT 12 (COUNT_MISMATCH)
        raise ValueError(f"Row count mismatch: {pre_count} vs {post_count}")

    print(f"Step 3 PASS: {post_count:,} rows written to {target_table}")
    print(f"Pipeline completed successfully (exit_code={exit_code})")

except ValueError:
    raise
except Exception as e:
    print(f"Step 3 FAIL: {str(e)}")
    exit_code = 8
    raise

# Fabric Spark: Player lifetime value with tier ranking
from pyspark.sql.functions import sum as spark_sum, count, datediff, \
    current_date, dense_rank, col
from pyspark.sql.window import Window

df_players = spark.table("silver.player_profiles")
df_txns = spark.table("silver.slot_transactions")

df_ltv = df_txns.groupBy("player_id").agg(
    spark_sum("coin_in").alias("lifetime_coin_in"),
    spark_sum("coin_out").alias("lifetime_coin_out"),
    count("*").alias("total_sessions"),
    spark_sum(col("coin_in") - col("coin_out")).alias("lifetime_net_revenue")
)

# Rank players by net revenue (replaces Teradata QUALIFY with DENSE_RANK)
window_spec = Window.orderBy(col("lifetime_net_revenue").desc())
df_ranked = df_ltv.withColumn("revenue_rank", dense_rank().over(window_spec))

df_result = df_ranked.join(df_players, "player_id") \
    .select("player_id", "player_name", "loyalty_tier",
            "lifetime_coin_in", "lifetime_net_revenue", "revenue_rank")

display(df_result.filter("revenue_rank <= 100"))

# Force V-ORDER rewrite on a migrated table
df = spark.table("gold.daily_slot_summary")
df.write \
    .mode("overwrite") \
    .option("vorder", "true") \
    .format("delta") \
    .saveAsTable("gold.daily_slot_summary")

print("V-ORDER rewrite complete - Direct Lake scans optimized")

# Utility: Auto-detect and report type mapping issues post-migration
from pyspark.sql.functions import col, min as spark_min, max as spark_max

def audit_type_mapping(table_name, checks):
    """Run type-mapping audit checks on a migrated table."""
    df = spark.table(table_name)
    print(f"Type Mapping Audit: {table_name}")
    print("=" * 60)

    for col_name, expected_type, check_fn in checks:
        actual_type = str(df.schema[col_name].dataType)
        type_ok = expected_type in actual_type
        range_result = check_fn(df, col_name) if check_fn else "N/A"
        status = "PASS" if type_ok else "FAIL"
        print(f"  [{status}] {col_name}: {actual_type} "
              f"(expected {expected_type}) | Range: {range_result}")

# Example: Audit slot_transactions
def check_range(df, col_name):
    row = df.select(spark_min(col(col_name)), spark_max(col(col_name))).collect()[0]
    return f"[{row[0]} .. {row[1]}]"

audit_type_mapping("bronze.slot_transactions", [
    ("coin_in",               "Decimal", check_range),
    ("coin_out",              "Decimal", check_range),
    ("transaction_timestamp", "Timestamp", check_range),
    ("machine_id",            "String", None),
])