If you’ve spent any time in Databricks, you’ve probably come across Autoloader, Structured Streaming, and Delta Live Tables (DLT).
And if you’re anything like me when I first started, you might have thought:
Why do we need three different things?
The short answer: They can work together, but they’re designed for different layers of your data ingestion and transformation pipeline.
1. Structured Streaming – The Core Engine
Think of Structured Streaming as the underlying engine for processing streaming and incremental data in Databricks. it’s part of Apache Spark.
What it does
Processes data in micro-batches or continuous mode from a source to a sink.
- Sources: Azure Event Hubs, Kinesis, cloud storage files, Delta tables, etc.
- Sinks: Delta tables, console output, Azure Event Hubs, etc.
You define a streaming query, and Spark keeps it running, processing new data as it arrives.
You can also run Structured Streaming in triggered batch mode instead of continuous mode.
Example: Process every 1 hour.
from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType # adjust to your payload
# 1) Define your Event Hubs connection string (use a secret in practice)
event_hub_connection = "<Event Hub connection string>"
# 2) Optional: specify starting position (latest) to avoid replay from beginning
starting_pos = """
{
"offset": "-1",
"seqNo": -1,
"enqueuedTime": null,
"isInclusive": false
}
"""
eh_reader = (spark.readStream
.format("eventhubs")
.option("eventhubs.connectionString", event_hub_connection)
.option("eventhubs.startingPosition", starting_pos)
)
raw_df = eh_reader.load()
# 3) Parse the payload: Event Hubs 'body' is binary; cast to string then parse JSON
json_schema = StructType([
StructField("id", StringType()),
StructField("name", StringType()),
StructField("value", DoubleType()),
StructField("timestamp", StringType())
])
parsed_df = (raw_df
.selectExpr("CAST(body AS STRING) AS message")
.withColumn("json", from_json(col("message"), json_schema))
.select("json.*") # or keep both 'message' and parsed columns if useful
.withColumn("__ingest_ts__", current_timestamp())
)
# 4) Write to Delta (MICRO-BATCH). Lower the trigger interval for near-real time.
query = (parsed_df.writeStream
.format("delta")
.option("checkpointLocation", "/mnt/checkpoints/eventhub") # must be stable & unique per stream
.outputMode("append")
.trigger(processingTime="10 seconds") # was "1 hour"; use 5–30s for low latency
.start("/mnt/delta/eventhub_data")
)
Limitations
- You must manage your own checkpoints by setting
checkpointLocation— without it, Spark won’t know where it left off. - Schema evolution requires extra config.
- If reading from cloud storage, you must handle file discovery logic yourself — unless you use Autoloader.
2. Databricks Autoloader – The Cloud File Ingestion Layer
Autoloader is built on top of Structured Streaming, but it solves a specific pain point:
Efficiently discovering and ingesting new files in cloud storage without rescanning the whole directory.
Without it, your job might waste compute by repeatedly scanning the entire folder.
What it does
Uses event notifications or directory listing mode to detect and load only new files.
- Formats: JSON, CSV, Avro, Parquet, ORC, text, etc.
- Sources: ADLS, Azure Blob, Amazon S3, Google Cloud Storage, Databricks File System (DBFS).
Schema evolution
Can automatically detect and handle new columns if you set cloudFiles.schemaEvolutionMode.
from pyspark.sql.functions import input_file_name, current_timestamp
raw_path = "/mnt/data/raw" # landing
bronze_path= "/mnt/delta/bronze" # Delta sink
chkpt_path = "/mnt/checkpoints/raw" # checkpoint
schema_loc = "/mnt/checkpoints/raw_schema" # schema tracking for Auto Loader
df = (spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "parquet")
# Track & evolve schema safely over time:
.option("cloudFiles.schemaLocation", schema_loc)
.option("cloudFiles.inferColumnTypes", "true")
.option("cloudFiles.schemaEvolutionMode", "addNewColumns")
# Optional but recommended at scale (requires storage notifications setup):
# .option("cloudFiles.useNotifications", "true")
# .option("cloudFiles.maxFilesPerTrigger", "1000")
.load(raw_path)
.withColumn("_source_file", input_file_name())
.withColumn("_ingest_ts", current_timestamp())
)
query = (df.writeStream
.format("delta")
.outputMode("append")
.option("checkpointLocation", chkpt_path)
# Allow new columns discovered by Auto Loader to land in Delta:
.option("mergeSchema", "true")
# Choose ONE trigger:
# .trigger(processingTime="10 seconds") # near-real-time streaming
# .trigger(availableNow=True) # backfill then stop (batch-like)
.start(bronze_path)
)
Limitations
- Still requires you to write streaming code.
- Only works with file-based sources.
- Event notification mode requires extra setup (e.g., Azure Event Grid).
- No built-in orchestration — transformations must be coded manually.
3. Delta Live Tables (DLT) – The Managed Pipeline Layer
Delta Live Tables is Databricks’ managed data pipeline framework.
Instead of writing complex streaming code, you define what you want in SQL or Python, and DLT takes care of how to run it.
What it does
- Automates ingestion (via Autoloader for files).
- Manages dependencies between tables.
- Handles retries, recovery, and orchestration.
- Enforces data quality rules.
- Runs in continuous or scheduled batch mode.
Example – Bronze, Silver, and Gold with data quality
import dlt
from pyspark.sql.functions import col, to_timestamp, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
# --- Assumptions (adjust to your data) ---
# - CSV has header row
# - Columns: order_id, region, amount, status, event_time (ISO8601)
bronze_schema = StructType([
StructField("order_id", StringType()),
StructField("region", StringType()),
StructField("amount", StringType()), # raw as string in CSV; cast later
StructField("status", StringType()),
StructField("event_time", StringType()), # raw as string; parse to timestamp
])
@dlt.table(
comment="Raw orders ingested from /mnt/raw/orders via Auto Loader (CSV)."
)
def bronze_orders():
return (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("cloudFiles.schemaLocation", "/mnt/checkpoints/orders_schema") # required for Auto Loader
.option("header", "true")
.schema(bronze_schema) # prefer explicit schema for CSV
.load("/mnt/raw/orders")
# Parse / type-cast early so downstream expectations are meaningful:
.withColumn("amount_d", col("amount").cast("double"))
.withColumn("event_ts", to_timestamp("event_time"))
.withColumn("_ingest_ts", current_timestamp())
)
# Data quality: drop bad rows rather than just record them (choose one style)
@dlt.expect_or_drop("positive_amount", "amount_d > 0")
@dlt.expect_or_drop("non_null_region", "region IS NOT NULL")
@dlt.table(
comment="Filtered confirmed orders with typed columns."
)
def silver_orders():
return (
dlt.read_stream("bronze_orders") # <--- stream read
.filter(col("status") == "confirmed")
.select("order_id", "region", "amount_d", "event_ts", "_ingest_ts")
)
# Streaming aggregation with watermark (recommended)
@dlt.table(
comment="Running sales by region (stateful streaming agg)."
)
def gold_sales_summary():
return (
dlt.read_stream("silver_orders")
# Watermark prevents unbounded state due to very late data
.withWatermark("event_ts", "1 day")
.groupBy("region")
.sum("amount_d")
.withColumnRenamed("sum(amount_d)", "total_amount")
)
4. Do They Support Callbacks?
Structured Streaming and Autoloader allows you to run custom code on each micro-batch of data as it arrives.
from delta.tables import DeltaTable
from pyspark.sql import functions as F
PROCESSED_PATH = "/mnt/delta/processed"
CALLBACK_LOG = "hive_metastore.logs.streaming_callback_log" # or UC path
# ensure log table exists once
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {CALLBACK_LOG} (
batch_id LONG,
started_at TIMESTAMP,
finished_at TIMESTAMP,
status STRING,
message STRING
) USING DELTA
""")
def process_batch(batch_df, batch_id):
start_ts = F.current_timestamp()
# 1) add audit columns
batch_df = (batch_df
.withColumn("__batch_id__", F.lit(batch_id))
.withColumn("__ingest_ts__", F.current_timestamp()))
# 2) write/upsert idempotently
# Assumption: 'id' uniquely identifies a row in the sink. Adjust as needed.
if DeltaTable.isDeltaTable(spark, PROCESSED_PATH):
tgt = DeltaTable.forPath(spark, PROCESSED_PATH)
(tgt.alias("t").merge(
batch_df.alias("s"),
"t.id = s.id" # <-- change to your natural/business key
).whenMatchedUpdateAll()
.whenNotMatchedInsertAll()
.execute())
else:
(batch_df.write
.format("delta")
.mode("overwrite")
.option("overwriteSchema", "true")
.save(PROCESSED_PATH))
# 3) record callback completion (even if retried, it’s just a log)
log_df = spark.createDataFrame(
[(int(batch_id),)], ["batch_id"]
).withColumn("started_at", start_ts)\
.withColumn("finished_at", F.current_timestamp())\
.withColumn("status", F.lit("SUCCESS"))\
.withColumn("message", F.lit("batch processed"))
(log_df.write
.format("delta").mode("append")
.saveAsTable(CALLBACK_LOG))
query = (df.writeStream
.foreachBatch(process_batch)
.option("checkpointLocation", "/mnt/checkpoints/processed_ckpt")
.start())
Delta Live Tables takes a different approach — it’s not designed for inline callbacks inside the pipeline code. Instead, you’d handle “after pipeline completion” logic using Databricks Workflows.
5. Key Takeaways
- Structured Streaming is the engine — you control it via code.
- Autoloader is a file loader — still uses Structured Streaming, but makes file ingestion faster and cheaper.
- Delta Live Tables is a managed service — orchestration, ingestion, transformations, and quality checks in one place.
- They are complementary, not competing.
- Both Structured Streaming and DLT can run continuously or on a schedule.