pyspark.md

Spark Configuration (Best Practices)

# Enable Fabric optimizations
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")
spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "true")

Reading Data

# Read CSV file
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("Files/bronze/data.csv")

# Read JSON file
df = spark.read.format("json").load("Files/bronze/data.json")

# Read Parquet file
df = spark.read.format("parquet").load("Files/bronze/data.parquet")

# Read Delta table
df = spark.read.table("my_delta_table")

# Read from SQL endpoint
df = spark.sql("SELECT * FROM lakehouse.my_table")

Writing Delta Tables

# Write DataFrame as managed Delta table
df.write.format("delta") \
    .mode("overwrite") \
    .saveAsTable("silver_customers")

# Write with partitioning
df.write.format("delta") \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .saveAsTable("silver_transactions")

# Append to existing table
df.write.format("delta") \
    .mode("append") \
    .saveAsTable("silver_events")

Delta Table Operations (CRUD)

# UPDATE
spark.sql("""
    UPDATE silver_customers
    SET status = 'active'
    WHERE last_login > '2024-01-01' -- Example date, adjust as needed
""")

# DELETE
spark.sql("""
    DELETE FROM silver_customers
    WHERE is_deleted = true
""")

# MERGE (Upsert)
spark.sql("""
    MERGE INTO silver_customers AS target
    USING staging_customers AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

Schema Definition

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType, DecimalType

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("name", StringType(), True),
    StructField("email", StringType(), True),
    StructField("amount", DecimalType(18, 2), True),
    StructField("created_at", TimestampType(), True)
])

df = spark.read.format("csv") \
    .schema(schema) \
    .option("header", "true") \
    .load("Files/bronze/customers.csv")

SQL Magic in Notebooks

%%sql
-- Query Delta table directly
SELECT 
    customer_id,
    COUNT(*) as order_count,
    SUM(amount) as total_amount
FROM gold_orders
GROUP BY customer_id
ORDER BY total_amount DESC
LIMIT 10

V-Order Optimization

# Enable V-Order for read optimization
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

Table Optimization

%%sql
-- Optimize table (compact small files)
OPTIMIZE silver_transactions

-- Optimize with Z-ordering on query columns
OPTIMIZE silver_transactions ZORDER BY (customer_id, transaction_date)

-- Vacuum old files (default 7 days retention)
VACUUM silver_transactions

-- Vacuum with custom retention
VACUUM silver_transactions RETAIN 168 HOURS

Incremental Load Pattern

from pyspark.sql.functions import col

# Get last processed watermark
last_watermark = spark.sql("""
    SELECT MAX(processed_timestamp) as watermark 
    FROM silver_orders
""").collect()[0]["watermark"]

# Load only new records
new_records = spark.read.format("delta") \
    .table("bronze_orders") \
    .filter(col("created_at") > last_watermark)

# Merge new records
new_records.createOrReplaceTempView("staging_orders")
spark.sql("""
    MERGE INTO silver_orders AS target
    USING staging_orders AS source
    ON target.order_id = source.order_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

SCD Type 2 Pattern

from pyspark.sql.functions import current_timestamp, lit

# Close existing records
spark.sql("""
    UPDATE dim_customer
    SET is_current = false, end_date = current_timestamp()
    WHERE customer_id IN (SELECT customer_id FROM staging_customer)
    AND is_current = true
""")

# Insert new versions
spark.sql("""
    INSERT INTO dim_customer
    SELECT 
        customer_id,
        name,
        email,
        address,
        current_timestamp() as start_date,
        null as end_date,
        true as is_current
    FROM staging_customer
""")