Spark on YARN to Databricks / Fabric Spark Migration¶

# Run this on your YARN Spark cluster to identify compatibility issues
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Check for deprecated API usage
print(f"Spark version: {spark.version}")
print(f"Scala version: {spark.sparkContext._jvm.scala.util.Properties.versionString()}")

# Check for legacy configurations
legacy_configs = [
    "spark.sql.legacy.timeParserPolicy",
    "spark.sql.legacy.createHiveTableByDefault",
    "spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation",
]
for config in legacy_configs:
    try:
        val = spark.conf.get(config)
        print(f"  {config} = {val}")
    except Exception:
        print(f"  {config} = not set")

# In a Fabric notebook, the code is the same PySpark
# Scheduling is configured in the Fabric UI or via REST API

# The notebook runs in a Fabric Spark pool with auto-scaling
# No spark-submit needed — the notebook IS the job

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Same PySpark code as before, just change paths
df = spark.read.format("delta").load(
    "abfss://silver@onelake.dfs.fabric.microsoft.com/warehouse/orders/"
)
# ... transformations ...
df.write.format("delta").mode("overwrite").save(
    "abfss://gold@onelake.dfs.fabric.microsoft.com/warehouse/daily_orders/"
)

# Method 1: Cluster libraries (installed on cluster start)
# Configure in cluster settings or via API:
# - PyPI packages: pandas, numpy, scikit-learn
# - Maven packages: com.mysql:mysql-connector-j:8.0.33
# - Custom JARs: uploaded to DBFS or Unity Catalog volumes

# Method 2: Notebook-scoped libraries
%pip install pandas==2.1.0 scikit-learn==1.3.0

# Method 3: Init scripts (for system-level dependencies)
# dbfs:/init-scripts/install-gdal.sh
#!/bin/bash
apt-get install -y gdal-bin libgdal-dev

# Method 4: Unity Catalog volumes (recommended for JARs)
# Upload JAR to volume, reference in cluster config

# Method 1: Fabric environment configuration
# Define Python packages and JARs in the Fabric environment UI

# Method 2: In-line installation in notebooks
%pip install pandas==2.1.0

# Method 3: Custom Spark properties
%%configure
{
    "conf": {
        "spark.jars": "abfss://libs@onelake.dfs.fabric.microsoft.com/custom-udf.jar"
    }
}

# BEFORE: Parquet on HDFS
df = spark.read.parquet("hdfs:///data/orders/")
df.write.mode("overwrite").parquet("hdfs:///data/daily_orders/")

# AFTER: Delta on ADLS Gen2
df = spark.read.format("delta").load("abfss://silver@storage.dfs.core.windows.net/orders/")
df.write.format("delta").mode("overwrite").save("abfss://gold@storage.dfs.core.windows.net/daily_orders/")

# MERGE (upsert) — replaces complex insert-overwrite patterns
from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "abfss://silver@storage.dfs.core.windows.net/orders/")
source = spark.read.format("delta").load("abfss://bronze@storage.dfs.core.windows.net/new_orders/")

target.alias("t").merge(
    source.alias("s"),
    "t.order_id = s.order_id"
).whenMatchedUpdate(set={
    "amount": "s.amount",
    "status": "s.status",
    "updated_at": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "order_id": "s.order_id",
    "customer_id": "s.customer_id",
    "amount": "s.amount",
    "status": "s.status",
    "created_at": "current_timestamp()",
    "updated_at": "current_timestamp()"
}).execute()

# BEFORE: Structured streaming on YARN reading from Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092") \
    .option("subscribe", "orders") \
    .load()

parsed = df.selectExpr("CAST(value AS STRING)").select(from_json("value", schema).alias("data")).select("data.*")

parsed.writeStream \
    .format("parquet") \
    .option("checkpointLocation", "hdfs:///checkpoints/orders/") \
    .option("path", "hdfs:///data/streaming_orders/") \
    .trigger(processingTime="1 minute") \
    .start()

# AFTER: Structured streaming on Databricks reading from Event Hubs (Kafka protocol)
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "mynamespace.servicebus.windows.net:9093") \
    .option("subscribe", "orders") \
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism", "PLAIN") \
    .option("kafka.sasl.jaas.config",
        'org.apache.kafka.common.security.plain.PlainLoginModule required '
        'username="$ConnectionString" '
        f'password="{dbutils.secrets.get("eventhubs", "connection-string")}";') \
    .load()

parsed = df.selectExpr("CAST(value AS STRING)").select(from_json("value", schema).alias("data")).select("data.*")

parsed.writeStream \
    .format("delta") \
    .option("checkpointLocation", "abfss://checkpoints@storage.dfs.core.windows.net/orders/") \
    .trigger(availableNow=True) \
    .toTable("silver.streaming_orders")