🔄 HDInsight Migration Guide¶

# Update Spark configuration for Azure storage
spark_conf = {
    "fs.azure.account.key.hadoopstorage.dfs.core.windows.net": "<storage-key>",
    "spark.hadoop.fs.defaultFS": "abfs://data@hadoopstorage.dfs.core.windows.net",
    "spark.sql.warehouse.dir": "abfs://data@hadoopstorage.dfs.core.windows.net/warehouse"
}

spark = SparkSession.builder \
    .appName("MigratedSparkJob") \
    .config(map=spark_conf) \
    .getOrCreate()

# Update data paths
input_path = "abfs://data@hadoopstorage.dfs.core.windows.net/raw/sales/"
output_path = "abfs://data@hadoopstorage.dfs.core.windows.net/processed/sales/"

df = spark.read.parquet(input_path)
df_processed = df.filter(col("amount") > 0)
df_processed.write.mode("overwrite").parquet(output_path)

# HDInsight Spark configuration
spark = SparkSession.builder \
    .appName("SalesETL") \
    .config("spark.hadoop.fs.azure.account.key.storage.dfs.core.windows.net", key) \
    .getOrCreate()

# Read from storage
df_raw = spark.read.parquet("abfs://data@storage.dfs.core.windows.net/raw/sales/")

# Transform
df_cleaned = df_raw \
    .filter(col("quantity") > 0) \
    .filter(col("price") > 0) \
    .withColumn("revenue", col("quantity") * col("price")) \
    .withColumn("year_month", date_format(col("transaction_date"), "yyyy-MM"))

# Write to warehouse
df_cleaned.write \
    .mode("overwrite") \
    .partitionBy("year_month") \
    .parquet("abfs://data@storage.dfs.core.windows.net/warehouse/sales/")

# Synapse Spark configuration (simplified)
# Authentication handled by Synapse workspace

from pyspark.sql.functions import col, sum as _sum, date_format
from delta.tables import DeltaTable

# Read from Data Lake (no manual credential configuration needed)
df_raw = spark.read.parquet("abfss://data@storage.dfs.core.windows.net/raw/sales/")

# Transform
df_cleaned = df_raw \
    .filter((col("quantity") > 0) & (col("price") > 0)) \
    .withColumn("revenue", col("quantity") * col("price")) \
    .withColumn("year_month", date_format(col("transaction_date"), "yyyy-MM"))

# Write to Delta Lake for ACID transactions
df_cleaned.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("year_month") \
    .save("abfss://data@storage.dfs.core.windows.net/warehouse/sales_delta/")

# Create table in shared metastore
spark.sql("""
    CREATE TABLE IF NOT EXISTS sales
    USING DELTA
    LOCATION 'abfss://data@storage.dfs.core.windows.net/warehouse/sales_delta/'
""")

# Optimize Delta table
spark.sql("OPTIMIZE sales")
spark.sql("VACUUM sales RETAIN 168 HOURS")

# Python script to automate Hive to Synapse migration

import pyodbc
from pyhive import hive

class HiveToSynapseMigration:
    def __init__(self, hive_host, synapse_connection_string):
        self.hive_conn = hive.Connection(host=hive_host, port=10000)
        self.synapse_conn = pyodbc.connect(synapse_connection_string)

    def get_hive_tables(self, database):
        """Get list of tables from Hive"""
        cursor = self.hive_conn.cursor()
        cursor.execute(f"USE {database}")
        cursor.execute("SHOW TABLES")
        return [row[0] for row in cursor.fetchall()]

    def get_table_schema(self, database, table):
        """Get table schema from Hive"""
        cursor = self.hive_conn.cursor()
        cursor.execute(f"DESCRIBE {database}.{table}")
        schema = cursor.fetchall()
        return schema

    def convert_hive_type_to_synapse(self, hive_type):
        """Convert Hive data type to Synapse SQL type"""
        type_mapping = {
            'string': 'VARCHAR(4000)',
            'int': 'INT',
            'bigint': 'BIGINT',
            'double': 'FLOAT',
            'decimal': 'DECIMAL(18,2)',
            'date': 'DATE',
            'timestamp': 'DATETIME2',
            'boolean': 'BIT'
        }
        return type_mapping.get(hive_type.lower(), 'VARCHAR(4000)')

    def create_synapse_table(self, database, table, schema, distribution_column=None):
        """Create table in Synapse SQL Pool"""
        cursor = self.synapse_conn.cursor()

        # Generate CREATE TABLE statement
        columns = []
        for col_name, col_type, _ in schema:
            if col_name and not col_name.startswith('#'):  # Skip partition info
                synapse_type = self.convert_hive_type_to_synapse(col_type)
                columns.append(f"    {col_name} {synapse_type}")

        distribution = f"DISTRIBUTION = HASH({distribution_column})" if distribution_column else "DISTRIBUTION = ROUND_ROBIN"

        create_statement = f"""
        CREATE TABLE dbo.{table}
        (
        {','.join(columns)}
        )
        WITH
        (
            {distribution},
            CLUSTERED COLUMNSTORE INDEX
        );
        """

        cursor.execute(create_statement)
        self.synapse_conn.commit()
        print(f"Created table: dbo.{table}")

    def export_and_load_data(self, hive_database, hive_table, export_path):
        """Export data from Hive and load to Synapse"""

        # Export from Hive to Parquet
        export_query = f"""
        INSERT OVERWRITE DIRECTORY '{export_path}'
        STORED AS PARQUET
        SELECT * FROM {hive_database}.{hive_table}
        """

        hive_cursor = self.hive_conn.cursor()
        hive_cursor.execute(export_query)

        # Load into Synapse using COPY
        copy_statement = f"""
        COPY INTO dbo.{hive_table}
        FROM '{export_path}/*.parquet'
        WITH
        (
            FILE_TYPE = 'PARQUET',
            CREDENTIAL = (IDENTITY = 'Managed Identity'),
            MAXERRORS = 0
        );
        """

        synapse_cursor = self.synapse_conn.cursor()
        synapse_cursor.execute(copy_statement)
        self.synapse_conn.commit()

        print(f"Loaded data into dbo.{hive_table}")

    def migrate_database(self, hive_database, export_base_path):
        """Migrate entire Hive database to Synapse"""
        tables = self.get_hive_tables(hive_database)

        for table in tables:
            print(f"Migrating table: {hive_database}.{table}")

            # Get schema
            schema = self.get_table_schema(hive_database, table)

            # Create table in Synapse
            self.create_synapse_table(hive_database, table, schema)

            # Export and load data
            export_path = f"{export_base_path}/{table}"
            self.export_and_load_data(hive_database, table, export_path)

        print(f"Migration of database {hive_database} completed!")

# Usage
migration = HiveToSynapseMigration(
    hive_host='headnode0',
    synapse_connection_string='DRIVER={ODBC Driver 17 for SQL Server};SERVER=synapse.sql.azuresynapse.net;DATABASE=pool1;UID=sqladmin;PWD=password'
)

migration.migrate_database('warehouse', 'abfs://export@datalake.dfs.core.windows.net/migration/')

# HDInsight Spark job
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum

# Manual configuration required
spark = SparkSession.builder \
    .appName("CustomerAnalytics") \
    .config("spark.hadoop.fs.azure.account.key.storage.dfs.core.windows.net", "<key>") \
    .config("spark.executor.memory", "8g") \
    .config("spark.executor.cores", "4") \
    .getOrCreate()

# Read data
df = spark.read.parquet("abfs://data@storage.dfs.core.windows.net/customers/")

# Process
result = df.groupBy("customer_segment") \
    .agg(_sum("revenue").alias("total_revenue"))

# Write
result.write.mode("overwrite").parquet("abfs://data@storage.dfs.core.windows.net/results/")

# Databricks notebook
# Simplified configuration - workspace handles authentication

from pyspark.sql.functions import col, sum as _sum
from delta.tables import DeltaTable

# Read from Delta Lake
df = spark.read.format("delta").load("/mnt/data/customers")

# Process with Delta Lake optimizations
result = df.groupBy("customer_segment") \
    .agg(_sum("revenue").alias("total_revenue"))

# Write to Delta Lake with ACID transactions
result.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save("/mnt/data/results/customer_revenue")

# Optimize Delta table
spark.sql("OPTIMIZE delta.`/mnt/data/results/customer_revenue`")
spark.sql("VACUUM delta.`/mnt/data/results/customer_revenue` RETAIN 168 HOURS")

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Load data
data = spark.read.parquet("abfs://data@storage.dfs.core.windows.net/training_data/")

# Feature engineering
assembler = VectorAssembler(
    inputCols=["age", "income", "total_purchases"],
    outputCol="features_raw"
)

scaler = StandardScaler(inputCol="features_raw", outputCol="features")

lr = LogisticRegression(featuresCol="features", labelCol="churn")

# Create pipeline
pipeline = Pipeline(stages=[assembler, scaler, lr])

# Train
train, test = data.randomSplit([0.8, 0.2], seed=42)
model = pipeline.fit(train)

# Evaluate
predictions = model.transform(test)
evaluator = BinaryClassificationEvaluator(labelCol="churn")
auc = evaluator.evaluate(predictions)

print(f"AUC: {auc}")

# Save model manually
model.write().overwrite().save("abfs://models@storage.dfs.core.windows.net/churn_model/")

import mlflow
import mlflow.spark
from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Enable MLflow autologging
mlflow.spark.autolog()

# Start MLflow run
with mlflow.start_run(run_name="churn_prediction_v1"):

    # Load data from Delta Lake
    data = spark.read.format("delta").load("/mnt/data/training_data")

    # Log parameters
    mlflow.log_param("data_version", "2024-01-15")
    mlflow.log_param("train_split", 0.8)

    # Feature engineering
    assembler = VectorAssembler(
        inputCols=["age", "income", "total_purchases"],
        outputCol="features_raw"
    )

    scaler = StandardScaler(inputCol="features_raw", outputCol="features")

    lr = LogisticRegression(
        featuresCol="features",
        labelCol="churn",
        maxIter=100,
        regParam=0.01
    )

    pipeline = Pipeline(stages=[assembler, scaler, lr])

    # Train
    train, test = data.randomSplit([0.8, 0.2], seed=42)
    model = pipeline.fit(train)

    # Evaluate
    predictions = model.transform(test)
    evaluator = BinaryClassificationEvaluator(labelCol="churn", metricName="areaUnderROC")
    auc = evaluator.evaluate(predictions)

    # Log metrics
    mlflow.log_metric("auc", auc)
    mlflow.log_metric("test_size", test.count())

    # Log model (automatic with autolog)
    print(f"Model logged to MLflow. AUC: {auc}")

# Register model
model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
mlflow.register_model(model_uri, "churn_prediction")

# Manual batch ETL pipeline
from pyspark.sql.functions import col, current_timestamp, to_date

# Bronze layer - raw ingestion
df_bronze = spark.read \
    .format("json") \
    .load("abfs://landing@storage.dfs.core.windows.net/events/")

df_bronze.write \
    .format("delta") \
    .mode("append") \
    .save("/mnt/data/bronze/events")

# Silver layer - cleaning and transformation
df_silver = spark.read.format("delta").load("/mnt/data/bronze/events") \
    .filter(col("event_type").isNotNull()) \
    .filter(col("user_id").isNotNull()) \
    .withColumn("event_date", to_date(col("event_timestamp")))

df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("event_date") \
    .save("/mnt/data/silver/events")

# Gold layer - business aggregations
df_gold = spark.read.format("delta").load("/mnt/data/silver/events") \
    .groupBy("user_id", "event_date") \
    .agg(count("*").alias("event_count"))

df_gold.write \
    .format("delta") \
    .mode("overwrite") \
    .save("/mnt/data/gold/daily_user_activity")

# Delta Live Tables pipeline (declarative ETL)
import dlt
from pyspark.sql.functions import col, current_timestamp, to_date, count

# Bronze layer - streaming ingestion
@dlt.table(
    comment="Raw events from landing zone",
    table_properties={"quality": "bronze"}
)
def bronze_events():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", "/mnt/checkpoint/bronze")
        .load("abfs://landing@storage.dfs.core.windows.net/events/")
    )

# Silver layer - with expectations (data quality)
@dlt.table(
    comment="Cleaned and validated events",
    table_properties={"quality": "silver"}
)
@dlt.expect_or_drop("valid_event_type", "event_type IS NOT NULL")
@dlt.expect_or_drop("valid_user_id", "user_id IS NOT NULL")
@dlt.expect("valid_timestamp", "event_timestamp > '2020-01-01'")
def silver_events():
    return (
        dlt.read_stream("bronze_events")
        .withColumn("event_date", to_date(col("event_timestamp")))
        .select("event_id", "user_id", "event_type", "event_date", "event_timestamp")
    )

# Gold layer - business aggregations
@dlt.table(
    comment="Daily user activity aggregation",
    table_properties={"quality": "gold"}
)
def gold_daily_user_activity():
    return (
        dlt.read("silver_events")
        .groupBy("user_id", "event_date")
        .agg(count("*").alias("event_count"))
    )

# Set up Unity Catalog for centralized governance

# Create catalog
spark.sql("CREATE CATALOG IF NOT EXISTS prod")

# Create schemas
spark.sql("CREATE SCHEMA IF NOT EXISTS prod.sales")
spark.sql("CREATE SCHEMA IF NOT EXISTS prod.marketing")
spark.sql("CREATE SCHEMA IF NOT EXISTS prod.ml_models")

# Grant permissions
spark.sql("""
    GRANT SELECT, MODIFY
    ON SCHEMA prod.sales
    TO `data-engineers@company.com`
""")

spark.sql("""
    GRANT SELECT
    ON SCHEMA prod.sales
    TO `analysts@company.com`
""")

# Register tables
spark.sql("""
    CREATE TABLE IF NOT EXISTS prod.sales.customers
    USING DELTA
    LOCATION '/mnt/data/customers'
    COMMENT 'Customer master data'
""")

# Add column-level metadata
spark.sql("""
    ALTER TABLE prod.sales.customers
    ALTER COLUMN email
    SET TAGS ('PII' = 'true', 'classification' = 'confidential')
""")

# Automated data validation script

from pyspark.sql import SparkSession
from pyspark.sql.functions import count, sum as _sum, col

class DataValidator:
    def __init__(self, spark):
        self.spark = spark

    def compare_row_counts(self, source_path, target_path, table_name):
        """Compare row counts between source and target"""
        source_count = self.spark.read.parquet(source_path).count()
        target_count = self.spark.read.format("delta").load(target_path).count()

        match = source_count == target_count
        print(f"Table: {table_name}")
        print(f"  Source count: {source_count}")
        print(f"  Target count: {target_count}")
        print(f"  Match: {match}")

        return match

    def compare_aggregations(self, source_df, target_df, agg_column, group_columns):
        """Compare aggregations between source and target"""
        source_agg = source_df.groupBy(*group_columns).agg(_sum(agg_column).alias("total"))
        target_agg = target_df.groupBy(*group_columns).agg(_sum(agg_column).alias("total"))

        # Join and compare
        comparison = source_agg.alias("source").join(
            target_agg.alias("target"),
            group_columns,
            "outer"
        ).select(
            *group_columns,
            col("source.total").alias("source_total"),
            col("target.total").alias("target_total")
        ).withColumn("match", col("source_total") == col("target_total"))

        mismatches = comparison.filter(col("match") == False)

        if mismatches.count() == 0:
            print("All aggregations match!")
            return True
        else:
            print("Mismatches found:")
            mismatches.show()
            return False

    def validate_schema(self, source_df, target_df):
        """Validate schema compatibility"""
        source_schema = set(source_df.schema.fieldNames())
        target_schema = set(target_df.schema.fieldNames())

        missing_in_target = source_schema - target_schema
        extra_in_target = target_schema - source_schema

        if missing_in_target:
            print(f"Columns missing in target: {missing_in_target}")

        if extra_in_target:
            print(f"Extra columns in target: {extra_in_target}")

        return len(missing_in_target) == 0

# Usage
validator = DataValidator(spark)

# Validate sales table migration
source_df = spark.read.parquet("abfs://hdinsight@storage.dfs.core.windows.net/sales/")
target_df = spark.read.format("delta").load("abfs://synapse@storage.dfs.core.windows.net/sales/")

validator.compare_row_counts(
    "abfs://hdinsight@storage.dfs.core.windows.net/sales/",
    "abfs://synapse@storage.dfs.core.windows.net/sales/",
    "sales"
)

validator.validate_schema(source_df, target_df)

validator.compare_aggregations(
    source_df,
    target_df,
    "revenue",
    ["product_category", "region"]
)

# Performance comparison script

import time
from datetime import datetime

def measure_query_performance(spark, query, platform_name):
    """Measure query execution time"""
    start_time = time.time()

    result = spark.sql(query)
    row_count = result.count()  # Action to trigger execution

    end_time = time.time()
    execution_time = end_time - start_time

    print(f"Platform: {platform_name}")
    print(f"  Rows returned: {row_count}")
    print(f"  Execution time: {execution_time:.2f} seconds")

    return {
        "platform": platform_name,
        "execution_time": execution_time,
        "row_count": row_count,
        "timestamp": datetime.now()
    }

# Test query
test_query = """
SELECT
    product_category,
    SUM(revenue) as total_revenue,
    COUNT(DISTINCT customer_id) as unique_customers
FROM sales
WHERE transaction_date >= '2024-01-01'
GROUP BY product_category
ORDER BY total_revenue DESC
"""

# Measure HDInsight performance
hdinsight_result = measure_query_performance(spark, test_query, "HDInsight")

# Measure Synapse performance
synapse_result = measure_query_performance(spark, test_query, "Synapse Spark Pool")

# Compare results
speedup = hdinsight_result["execution_time"] / synapse_result["execution_time"]
print(f"\nSpeedup: {speedup:.2f}x")