Tutorial: Migrate HDFS Data to ADLS Gen2¶

# Cell 1: Configure ADLS access
# (If using managed identity, this is automatic on Databricks)
# If using account key:
spark.conf.set(
    "fs.azure.account.key.hadoopmigrationlake.dfs.core.windows.net",
    dbutils.secrets.get(scope="migration", key="storage-key")
)

# Cell 2: List tables in raw zone
tables = dbutils.fs.ls("abfss://raw@hadoopmigrationlake.dfs.core.windows.net/hive/warehouse/")
for t in tables:
    print(f"{t.name:40s} {t.size:>15,} bytes")

# Cell 3: Convert regions table (smallest)
source_path = "abfss://raw@hadoopmigrationlake.dfs.core.windows.net/hive/warehouse/regions"
target_path = "abfss://silver@hadoopmigrationlake.dfs.core.windows.net/regions"

# Read source (auto-detect format: ORC or Parquet)
try:
    df = spark.read.format("orc").load(source_path)
    source_format = "ORC"
except:
    df = spark.read.format("parquet").load(source_path)
    source_format = "Parquet"

print(f"Source format: {source_format}")
print(f"Row count: {df.count():,}")
print(f"Schema:")
df.printSchema()

# Write as Delta
df.write.format("delta") \
    .mode("overwrite") \
    .save(target_path)

print(f"Successfully converted {source_format} to Delta at {target_path}")

# Cell 4: Batch convert all tables
from pyspark.sql import functions as F

results = []

for table_dir in tables:
    table_name = table_dir.name.rstrip("/")
    source_path = table_dir.path
    target_path = f"abfss://silver@hadoopmigrationlake.dfs.core.windows.net/{table_name}"

    try:
        # Attempt to read as ORC first, then Parquet
        try:
            df = spark.read.format("orc").load(source_path)
            fmt = "ORC"
        except:
            df = spark.read.format("parquet").load(source_path)
            fmt = "Parquet"

        row_count = df.count()

        # Detect partition columns (Hive-style partition directories)
        partition_cols = [c for c in df.columns
                         if c in ("year", "month", "day", "date", "region", "country")]

        # Write as Delta
        writer = df.write.format("delta").mode("overwrite")
        if partition_cols:
            writer = writer.partitionBy(*partition_cols)
        writer.save(target_path)

        results.append((table_name, fmt, row_count, "SUCCESS", ", ".join(partition_cols) or "none"))
        print(f"  [OK] {table_name}: {row_count:,} rows ({fmt} -> Delta)")

    except Exception as e:
        results.append((table_name, "unknown", 0, f"FAILED: {str(e)[:100]}", ""))
        print(f"  [FAIL] {table_name}: {str(e)[:100]}")

# Display summary
results_df = spark.createDataFrame(results, ["table", "source_format", "rows", "status", "partitions"])
display(results_df)

# Cell 6: Register all Delta tables in Unity Catalog
for table_dir in tables:
    table_name = table_dir.name.rstrip("/")
    target_path = f"abfss://silver@hadoopmigrationlake.dfs.core.windows.net/{table_name}"

    try:
        spark.sql(f"""
            CREATE TABLE IF NOT EXISTS migration.silver.{table_name}
            USING DELTA
            LOCATION '{target_path}'
        """)
        print(f"  [OK] Registered migration.silver.{table_name}")
    except Exception as e:
        print(f"  [FAIL] {table_name}: {str(e)[:100]}")

# Cell 9: Compare row counts between raw (source format) and silver (Delta)
validation = []

for table_dir in tables:
    table_name = table_dir.name.rstrip("/")
    raw_path = table_dir.path
    silver_path = f"abfss://silver@hadoopmigrationlake.dfs.core.windows.net/{table_name}"

    try:
        # Count in raw zone
        try:
            raw_count = spark.read.format("orc").load(raw_path).count()
        except:
            raw_count = spark.read.format("parquet").load(raw_path).count()

        # Count in silver zone (Delta)
        silver_count = spark.read.format("delta").load(silver_path).count()

        match = raw_count == silver_count
        validation.append((table_name, raw_count, silver_count, match))

    except Exception as e:
        validation.append((table_name, -1, -1, False))

val_df = spark.createDataFrame(validation, ["table", "raw_count", "delta_count", "match"])
display(val_df)

# Assert all match
failures = val_df.filter("match = false")
if failures.count() > 0:
    print("VALIDATION FAILED for the following tables:")
    display(failures)
else:
    print("ALL TABLES VALIDATED SUCCESSFULLY")

# Cell 10: Aggregate validation for a key table
table_name = "orders"  # Replace with your largest/most important table
raw_path = f"abfss://raw@hadoopmigrationlake.dfs.core.windows.net/hive/warehouse/{table_name}"
silver_path = f"abfss://silver@hadoopmigrationlake.dfs.core.windows.net/{table_name}"

try:
    raw_df = spark.read.format("orc").load(raw_path)
except:
    raw_df = spark.read.format("parquet").load(raw_path)

silver_df = spark.read.format("delta").load(silver_path)

# Compare key metrics
for df, label in [(raw_df, "RAW"), (silver_df, "DELTA")]:
    stats = df.agg(
        F.count("*").alias("row_count"),
        F.sum("amount").alias("total_amount"),
        F.countDistinct("customer_id").alias("unique_customers"),
        F.min("order_date").alias("min_date"),
        F.max("order_date").alias("max_date")
    ).collect()[0]

    print(f"\n{label}:")
    print(f"  Row count:        {stats['row_count']:,}")
    print(f"  Total amount:     {stats['total_amount']:,.2f}")
    print(f"  Unique customers: {stats['unique_customers']:,}")
    print(f"  Date range:       {stats['min_date']} to {stats['max_date']}")

# Cell 11: Schema comparison
table_name = "orders"
raw_path = f"abfss://raw@hadoopmigrationlake.dfs.core.windows.net/hive/warehouse/{table_name}"
silver_path = f"abfss://silver@hadoopmigrationlake.dfs.core.windows.net/{table_name}"

try:
    raw_schema = spark.read.format("orc").load(raw_path).schema
except:
    raw_schema = spark.read.format("parquet").load(raw_path).schema

silver_schema = spark.read.format("delta").load(silver_path).schema

# Compare field by field
print(f"{'Column':<30} {'Raw Type':<20} {'Delta Type':<20} {'Match'}")
print("-" * 90)
for raw_field in raw_schema.fields:
    delta_field = silver_schema[raw_field.name] if raw_field.name in silver_schema.fieldNames() else None
    if delta_field:
        match = str(raw_field.dataType) == str(delta_field.dataType)
        print(f"{raw_field.name:<30} {str(raw_field.dataType):<20} {str(delta_field.dataType):<20} {'OK' if match else 'MISMATCH'}")
    else:
        print(f"{raw_field.name:<30} {str(raw_field.dataType):<20} {'MISSING':<20} FAIL")