Reading from Any Source & Writing to Microsoft Fabric Lakehouse with PySpark

PySpark Oct 2, 2026 · 12 min read

Microsoft Fabric Lakehouse brings together the best of data lakes and data warehouses. Combined with PySpark, you can build robust ETL pipelines that read from virtually any source and land clean, typed data into Delta Lake tables. This guide walks through the entire process.

Prerequisites

1. Reading CSV files

PySpark handles CSV natively with schema inference and configurable options for headers, delimiters, and null handling.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

df_csv = (spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .option("nullValue", "")
    .csv("Files/raw/customers.csv"))

df_csv.printSchema()
df_csv.show(5)

For production workloads, define the schema explicitly instead of relying on inference — it's faster and prevents type mismatches across file batches.

2. Reading JSON files

JSON files, including nested structures, are handled naturally by Spark's DataFrame API.

df_json = (spark.read
    .option("multiline", "true")
    .json("Files/raw/orders.json"))

# Flatten nested structs
from pyspark.sql.functions import col
df_flat = df_json.select(
    col("order_id"),
    col("customer.name").alias("customer_name"),
    col("customer.email").alias("customer_email"),
    col("amount"),
    col("created_at")
)

3. Reading Parquet files

Parquet is the most efficient format for Lakehouse workloads — it preserves schema, supports predicate pushdown, and compresses well.

df_parquet = spark.read.parquet("Files/raw/events/")

# Parquet retains types, so no inference needed
print(f"Rows: {df_parquet.count()}")
print(f"Columns: {len(df_parquet.columns)}")

4. Reading from REST APIs

For API sources, fetch the data with requests and convert to a DataFrame.

import requests
import json

response = requests.get("https://api.example.com/v1/products",
    headers={"Authorization": "Bearer YOUR_TOKEN"})

data = response.json()["results"]
df_api = spark.createDataFrame(data)

5. Writing to Fabric Lakehouse (Delta Lake)

The key step — writing DataFrames as Delta tables inside the Lakehouse. Fabric manages the metastore automatically.

# Write as a managed Delta table
df_csv.write \
    .mode("overwrite") \
    .format("delta") \
    .saveAsTable("lakehouse.customers")

# Append new data
df_json.write \
    .mode("append") \
    .format("delta") \
    .saveAsTable("lakehouse.orders")

Use mode("overwrite") for full refreshes and mode("append") for incremental loads. For upserts, use Delta Lake's MERGE INTO operation.

6. Schema evolution

When source schemas change over time, enable automatic schema merging:

df_new.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .format("delta") \
    .saveAsTable("lakehouse.customers")

7. Putting it all together

Here's a complete pipeline pattern that reads from multiple sources and writes to a single Lakehouse table:

from pyspark.sql.functions import current_timestamp, lit

def ingest_to_lakehouse(df, table_name, source_name, mode="append"):
    """Standard ingestion with audit columns."""
    df_enriched = (df
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source", lit(source_name)))

    df_enriched.write \
        .mode(mode) \
        .option("mergeSchema", "true") \
        .format("delta") \
        .saveAsTable(f"lakehouse.{table_name}")

    print(f"✅ {df.count()} rows → {table_name} from {source_name}")

# Run
ingest_to_lakehouse(df_csv, "customers", "csv_upload")
ingest_to_lakehouse(df_api, "products", "rest_api")
ingest_to_lakehouse(df_parquet, "events", "parquet_archive")

Key takeaways