Study guide
7.4 Structured Streaming Fundamentals & Azure Event Hubs / Kafka Ingestion
Microsoft Certified: Azure Databricks Data Engineer Associate DP-750
Track your progress as you read
Finished sections get a check mark.
1 Chapter 1: Azure Databricks Architecture & Compute Selection
1.1 DP-750 Exam Overview & Azure Databricks Lakehouse Architecture 1.2 Compute Types & Workload Matching 1.3 Cluster Access Modes, Standard/Dedicated, & Serverless Compute
2 Chapter 2: Compute Configuration, Performance & Cost Management
2.1 Autoscaling, Databricks Pools, & Node Type Selection 2.2 Photon Vectorized Engine & Cost Governance with Cluster Policies 2.3 Library Management: Workspace, Compute-Scoped, Notebook, & Init Scripts 2.4 Auto-Termination, Budget Alerts, & Tagging for Cost Allocation 2.5 Databricks Runtime & ML Runtime Selection, and Compute Access Permissions
3 Chapter 3: Unity Catalog Storage Infrastructure & Workspace Setup
3.1 Unity Catalog 3-Level Namespace & Metastore Architecture 3.2 Azure Managed Identity, Storage Credentials, & External Locations 3.3 Managed Tables, External Tables, & Unity Catalog Volumes 3.4 Unity Catalog Naming Conventions, Environment Isolation, & Data Granularity 3.5 Lakehouse Federation: Connections, Foreign Catalogs, & Query Pushdown
4 Chapter 4: Unity Catalog Privilege Model & Access Control
4.1 Hierarchical Privilege Model & Securable Objects 4.2 Account-Level Identities, SCIM Sync, & Group Management 4.3 Object Ownership, Ownership Transfer, & Administrative Delegation
5 Chapter 5: Advanced Data Security, Masking & ABAC
5.1 Dynamic Row Filters & Column Masking Implementation 5.2 Attribute-Based Access Control (ABAC) with Unity Catalog Tags 5.3 Data Isolation across Workspaces & Multiple Metastore Patterns 5.4 Secure Data Sharing within Azure Databricks
6 Chapter 6: Governance, Auditing, Lineage & Delta Sharing
6.1 Automated Data Lineage Tracking in Catalog Explorer 6.2 Unity Catalog System Tables: Audit, Billing, and Access Logs 6.3 Data Retention Policies, Time Travel Security, & Tag Governance 6.4 Delta Sharing Protocols: Open Sharing vs. Databricks-to-Databricks 6.5 Data Discovery: Table and Column Descriptions, and AI/BI Genie Instructions
7 Chapter 7: Ingestion with Auto Loader, Batch & Structured Streaming
7.1 Auto Loader (cloudFiles) Architecture, File Notification vs. Directory Listing 7.2 Auto Loader Schema Inference, Evolution, & Rescued Data Column 7.3 Batch Ingestion with COPY INTO, CTAS, & Multipart Files 7.4 Structured Streaming Fundamentals & Azure Event Hubs / Kafka Ingestion 7.5 Streaming Watermarking, Triggers, & Output Modes 7.6 Choosing an Ingestion Tool: Lakeflow Connect, Notebooks, & Azure Data Factory
8 Chapter 8: Delta Lake Core Mechanics, ACID & Table Maintenance
8.1 Delta Lake Architecture, Parquet Files, & the _delta_log Transaction Log 8.2 ACID Transactions, Concurrency Control, & Conflict Resolution 8.3 Compaction (OPTIMIZE), Z-Ordering, & Liquid Clustering (CLUSTER BY) 8.4 Data Retention, Snapshot Isolation, & VACUUM Operations 8.5 Delta Change Data Feed (CDF) & MERGE INTO Upsert Patterns 8.6 Choosing a Table Format: Delta, Parquet, CSV, JSON, & Apache Iceberg
9 Chapter 9: Data Cleansing, Transformation & Medallion Architecture
9.1 The Medallion Lakehouse Pattern: Bronze, Silver, and Gold Layer Design 9.2 Data Cleansing: Deduplication, Null Handling, & Type Casting in PySpark/SQL 9.3 Complex Transformations: Higher-Order Functions, Arrays, Structs, & JSON Parsing 9.4 Relational Joins, Window Functions, Set Operators, & Pivoting 9.5 Enforcing Data Quality: Check Constraints & Table Invariants 9.6 Data Profiling: Summary Statistics & Distribution Assessment
10 Chapter 10: Declarative Pipelines, Lakeflow Spark & Delta Live Tables
10.1 Declarative Pipeline Fundamentals: Streaming Tables vs. Materialized Views 10.2 Data Quality Expectations: expect, expect or drop, & expect or fail 10.3 Change Data Capture Ingestion with AUTO CDC / APPLY CHANGES INTO 10.4 Pipeline Execution Modes, Serverless DLT, & Event Logs
11 Chapter 11: Workflow Orchestration with Lakeflow Jobs
11.1 Multi-Task DAG Design in Lakeflow Jobs & Task Type Selection 11.2 Task Dependencies, Parameters, Task Values, & Dynamic Variables 11.3 Triggers (Scheduled, File Arrival, Continuous), Retries, & Timeouts 11.4 Notifications, Alerting (Email, Slack, Webhooks), & Failure Handling
12 Chapter 12: CI/CD, Git Integration & Databricks Asset Bundles
12.1 Databricks Git Folders & Source Control Integration 12.2 Databricks Asset Bundles (DAB) Architecture & databricks.yml Structure 12.3 Multi-Environment Deployment (Dev / Staging / Prod) & Target Overrides 12.4 Automated Testing, Unit Testing PySpark, & CI/CD Pipelines
13 Chapter 13: Workload Monitoring, Troubleshooting, & Performance Tuning
13.1 Spark UI Mastery: Query Execution DAG, Stages, Tasks, & Skew Analysis 13.2 Troubleshooting Common Failures: OOM, Disk Spill, Data Skew, & Network Timeouts 13.3 Performance Tuning: Adaptive Query Execution (AQE), Caching, & Broadcast Joins 13.4 Workspace Monitoring with Azure Monitor, Diagnostic Settings, & Log Analytics
7.4 Structured Streaming Fundamentals & Azure Event Hubs / Kafka Ingestion
Key Takeaways Explain This Topic AI
Apache Spark Structured Streaming processes unbounded streaming data as an append-only table, providing end-to-end fault tolerance and exactly-once processing guarantees.
Azure Event Hubs provides a native Apache Kafka-compatible endpoint, enabling seamless ingestion into Azure Databricks using the standard Spark Kafka streaming format ('kafka').
Authentication between Azure Databricks and Azure Event Hubs / Kafka is securely established via SASL_SSL and JAAS configurations referencing Databricks Secret Scopes.
Kafka and Event Hubs sources emit records with standard binary schema columns ('key', 'value', 'topic', 'partition', 'offset', 'timestamp'), requiring explicit deserialization via functions like 'from_json()'.
Offset management parameters ('startingOffsets', 'maxOffsetsPerTrigger', 'minOffsetsPerTrigger') regulate initial stream placement and enforce rate limiting to prevent consumer lag spikes.
Last updated: August 2026
7.4 Structured Streaming Fundamentals & Azure Event Hubs / Kafka Ingestion
DP-750 Exam Focus: Understand the core architecture of Spark Structured Streaming (the unbounded table model, readStream , writeStream , and checkpointing). Master configuring real-time ingestion from Azure Event Hubs using the Kafka interface, securing connection credentials via Databricks Secret Scopes , deserializing binary payloads ( key , value ) with from_json() , and managing offsets and rate limits ( maxOffsetsPerTrigger ).
1. Structured Streaming Processing Model
Apache Spark Structured Streaming is a scalable and fault-tolerant stream processing engine built upon the Catalyst query optimization and Photon execution engines.
Rather than treating streaming data as discrete low-level RDD batches (the legacy DStream model), Structured Streaming models a real-time data stream as an Unbounded Table to which incoming records are continuously appended. +-----------------------------------------------------------------------------------+
- | STRUCTURED STREAMING UNBOUNDED TABLE MODEL |
- +-----------------------------------------------------------------------------------+
- | |
- | Time Incoming Stream Records Unbounded Input Table |
- | ---- ----------------------- --------------------- |
- | t0: [Row 1, Row 2] ---> [Row 1, Row 2] |
- | t1: [Row 3] ---> [Row 1, Row 2, Row 3] |
- | t2: [Row 4, Row 5] ---> [Row 1, Row 2, Row 3, Row 4, Row 5]|
- | |
- | | |
- | v |
- | [ Incremental Query Execution Plan ] |
- | - Catalyst Logical Optimization |
- | - State Store Management (RocksDB) |
- | | |
- | v |
- | [ Result Table / Target Delta Sink ] |
- +-----------------------------------------------------------------------------------+
- The readStream and writeStream Pipeline Lifecycle
Every Structured Streaming query follows a standardized programmatic pattern:
spark.readStream : Defines the streaming DataFrame reader attached to an unbounded streaming source (e.g., Kafka, Event Hubs, Auto Loader, or Delta Lake table).
Transformations: Stateless transformations (projections, filters, column additions) or stateful operations (windowed aggregations, stream-stream joins).
df.writeStream : Configures the output sink, output mode ( append , complete , update ), trigger interval, and mandatory checkpointLocation .
.start() : Initiates the background streaming execution thread and returns a StreamingQuery control handle.
2. Ingesting from Azure Event Hubs via Kafka Interface
Microsoft Azure Event Hubs exposes an Apache Kafka 1.0+ compatible endpoint , allowing Azure Databricks to ingest real-time events using the built-in, highly optimized Spark Kafka connector ( format("kafka") ). This avoids third-party JAR dependencies and leverages native Spark streaming optimizations. AZURE EVENT HUBS INGESTION FLOW
- +-------------------------------------------------------------------+
- | Azure Event Hubs Namespace: eh-telemetry-prod.servicebus.windows.net|
- | Event Hub (Topic): iot-device-events |
- +-------------------------------------------------------------------+
- |
- | (SASL_SSL / Port 9093)
- v
- +-------------------------------------------------------------------+
- | Azure Databricks Cluster (Spark Structured Streaming) |
- | - format("kafka") |
- | - dbutils.secrets.get("azure-eh", "connection-string") |
- | - Kafka JAAS Configuration |
- +-------------------------------------------------------------------+
- |
- v
- +-------------------------------------------------------------------+
- | Bronze Delta Lake Table: bronze.iot.telemetry_raw |
- +-------------------------------------------------------------------+
- Security & Connection Configuration
Connecting securely to Azure Event Hubs requires authenticating via SASL_SSL with a Shared Access Signature (SAS) connection string stored securely in a Databricks Secret Scope (backed by Azure Key Vault): # Step 1: Retrieve Event Hubs connection details from Databricks Secret Scope
- eh_namespace = "eh-telemetry-prod.servicebus.windows.net:9093"
- eh_topic = "iot-device-events"
- eh_conn_str = dbutils.secrets.get(scope="azure-keyvault-scope", key="eventhubs-conn-string")
- # Step 2: Construct the Kafka JAAS configuration string
- # Note: $ConnectionString is the required Kafka username for Azure Event Hubs
jaas_config = f'org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{eh_conn_str}";'
- # Step 3: Configure Spark Structured Streaming readStream
- df_eventhubs_raw = (spark.readStream
- .format("kafka")
- .option("kafka.bootstrap.servers", eh_namespace)
- .option("subscribe", eh_topic)
- .option("kafka.security.protocol", "SASL_SSL")
- .option("kafka.sasl.mechanism", "PLAIN")
- .option("kafka.sasl.jaas.config", jaas_config)
- .option("startingOffsets", "latest") # Options: 'earliest', 'latest', or JSON offset map
- .option("maxOffsetsPerTrigger", 10000) # Rate limiting: max messages per micro-batch
- .option("failOnDataLoss", "false") # Resilient to Event Hubs message retention truncation
- .load())
Important Exam Fact: When connecting to Azure Event Hubs using the Kafka protocol, the username in the JAAS config is always literally "$ConnectionString" , and the password is the complete Event Hubs Primary Connection String.
3. Message Deserialization & Binary Payload Processing
When reading from Kafka or Azure Event Hubs, Spark produces a DataFrame with a fixed binary schema:
Column Name Data Type Description
key BINARY Optional partition routing key emitted by producer.
value BINARY The primary message payload containing the raw JSON/Avro/byte data.
topic STRING The Event Hub / Kafka topic name from which the record was read.
partition INTEGER Specific partition ID (e.g., 0 , 1 , 2 ...).
offset LONG Sequential 64-bit integer tracking message position within partition.
timestamp TIMESTAMP Enqueue timestamp assigned by producer or Event Hubs broker.
timestampType INTEGER Type of timestamp ( 0 = CreateTime, 1 = LogAppendTime).
Deserializing JSON Payloads
Because the value column is stored as raw binary bytes, data engineers must cast it to STRING and apply from_json() with a predefined StructType schema to unpack the attributes into structured columns: from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType, TimestampType
- # Define payload schema
- payload_schema = StructType([
- StructField("device_id", StringType(), True),
- StructField("temperature", DoubleType(), True),
- StructField("humidity", DoubleType(), True),
- StructField("reading_time", TimestampType(), True)
- ])
- # Unpack binary payload and extract metadata
- df_parsed = (df_eventhubs_raw
- .select(
- col("topic"),
- col("partition"),
- col("offset"),
- col("timestamp").alias("enqueued_time"),
- from_json(col("value").cast("string"), payload_schema).alias("data")
- )
- .select(
- "topic",
- "partition",
- "offset",
- "enqueued_time",
- "data.*"
- ))
- 4. Offset Management, Rate Limiting, & Backpressure
- Offset Configurations on Startup
startingOffsets = 'latest' : Ingests only new messages that arrive after the query starts. Historical messages currently in the Event Hub partitions are skipped.
startingOffsets = 'earliest' : Ingests all available historical messages retained within the Event Hub's retention window (e.g., past 1 to 7 days), starting from offset 0 .
Specific Offset Map: Ingests from exact per-partition offsets provided in a JSON string (e.g., {"iot-device-events":{"0":14200,"1":18950}} ).
Exam Note: startingOffsets only takes effect when the stream starts for the very first time without an existing checkpoint directory. If a checkpoint directory already exists, Spark resumes from the persisted offsets in the checkpoint, ignoring startingOffsets .
Backpressure & Rate Limiting
maxOffsetsPerTrigger : Restricts the maximum number of records read across all partitions in a single micro-batch. Setting maxOffsetsPerTrigger = 50000 prevents large backlogs from causing driver out-of-memory errors and maintains predictable batch durations.
minOffsetsPerTrigger : Specifies a minimum number of records required before triggering a micro-batch, avoiding micro-batches on low-volume streams.
5. Checkpointing & Exactly-Once Ingestion Pipeline
To achieve end-to-end exactly-once guarantees, Structured Streaming pairs offset tracking with Delta Lake's ACID transaction log: # Production Pipeline: Writing parsed stream to Bronze Delta Table
- query = (df_parsed.writeStream
- .format("delta")
- .outputMode("append")
- .option("checkpointLocation", "abfss://checkpoints@myadls.dfs.core.windows.net/iot_stream_cp")
- .trigger(processingTime="10 seconds")
- .toTable("bronze.iot.telemetry_events"))
- Checkpoint Anatomy & State Recovery
offsets/ : Contains the starting and ending partition offsets for each micro-batch (e.g., Partition 0: offsets 100-200).
commits/ : Contains an atomic commit marker written only after the target Delta Lake table successfully commits the batch.
Failure Recovery Protocol: If a cluster node crashes mid-batch, upon restart the driver inspects commits/ . If the last batch in offsets/ lacks a matching entry in commits/ , Spark re-executes that exact offset range. Because Delta Lake uses transactional idempotence, replaying the micro-batch overwrites uncommitted data, ensuring zero duplicates and zero data loss.