Best Practices: Hadoop to Azure Migration¶

# reconciliation.py — Daily comparison of Hadoop vs Azure outputs

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.getOrCreate()

def reconcile_table(hadoop_path, azure_path, key_columns, metric_columns, table_name):
    """Compare Hadoop and Azure outputs for a single table."""

    hadoop_df = spark.read.parquet(hadoop_path)
    azure_df = spark.read.format("delta").load(azure_path)

    results = {
        "table": table_name,
        "hadoop_rows": hadoop_df.count(),
        "azure_rows": azure_df.count(),
        "row_match": False,
        "metric_matches": {},
        "mismatched_keys": []
    }

    # Row count comparison
    results["row_match"] = results["hadoop_rows"] == results["azure_rows"]

    # Aggregate metric comparison
    for col in metric_columns:
        h_val = hadoop_df.agg(F.sum(col)).collect()[0][0]
        a_val = azure_df.agg(F.sum(col)).collect()[0][0]
        # Allow 0.01% tolerance for floating-point differences
        tolerance = abs(h_val * 0.0001) if h_val else 0.01
        results["metric_matches"][col] = abs(h_val - a_val) < tolerance

    # Key-level comparison (find mismatched rows)
    hadoop_keys = hadoop_df.select(*key_columns).distinct()
    azure_keys = azure_df.select(*key_columns).distinct()

    missing_in_azure = hadoop_keys.subtract(azure_keys)
    missing_in_hadoop = azure_keys.subtract(hadoop_keys)

    results["missing_in_azure"] = missing_in_azure.count()
    results["missing_in_hadoop"] = missing_in_hadoop.count()

    return results


# Run reconciliation for all migrated tables
tables_to_reconcile = [
    {
        "name": "daily_revenue",
        "hadoop_path": "hdfs:///analytics/daily_revenue/",
        "azure_path": "abfss://gold@storage.dfs.core.windows.net/daily_revenue/",
        "key_columns": ["report_date"],
        "metric_columns": ["total_revenue", "total_orders"]
    },
    {
        "name": "customer_360",
        "hadoop_path": "hdfs:///analytics/customer_360/",
        "azure_path": "abfss://gold@storage.dfs.core.windows.net/customer_360/",
        "key_columns": ["customer_id"],
        "metric_columns": ["lifetime_revenue", "lifetime_orders"]
    }
]

for table_config in tables_to_reconcile:
    result = reconcile_table(**table_config)
    print(f"\n{'='*60}")
    print(f"Table: {result['table']}")
    print(f"  Row count:  Hadoop={result['hadoop_rows']:,}  Azure={result['azure_rows']:,}  Match={result['row_match']}")
    for metric, match in result["metric_matches"].items():
        print(f"  {metric}: Match={match}")
    print(f"  Missing in Azure: {result['missing_in_azure']}")
    print(f"  Extra in Azure: {result['missing_in_hadoop']}")

# Archive critical HDFS data to ADLS Archive tier for compliance
azcopy copy \
  "/mnt/hdfs-export/compliance-data/" \
  "https://archivestorage.dfs.core.windows.net/compliance/?<SAS>" \
  --recursive \
  --put-md5

# Set lifecycle policy for automatic deletion after retention period
az storage account management-policy create \
  --account-name archivestorage \
  --policy '{
    "rules": [{
      "name": "compliance-retention",
      "type": "Lifecycle",
      "definition": {
        "filters": {"blobTypes": ["blockBlob"], "prefixMatch": ["compliance/"]},
        "actions": {"baseBlob": {"delete": {"daysAfterModificationGreaterThan": 2555}}}
      }
    }]
  }'