Study guide
7.5 Streaming Watermarking, Triggers, & Output Modes
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.5 Streaming Watermarking, Triggers, & Output Modes
Key Takeaways Explain This Topic AI
Watermarking with 'withWatermark(timeColumn, delayThreshold)' establishes an event-time moving boundary that limits how long the engine retains state for late-arriving records in stateful aggregations and joins.
Stateful stream-stream joins require defined watermarks on both input streams alongside an event-time interval constraint to prevent unbounded state memory growth in RocksDB.
Structured Streaming triggers control execution cadence: 'Trigger.AvailableNow()' executes all outstanding data in cost-effective micro-batches and terminates, 'Trigger.ProcessingTime()' sets periodic intervals, and 'Trigger.Continuous()' enables low-latency execution.
Output modes define write semantics: 'Append' (default; emits only finalized rows; mandatory for append-only sinks), 'Complete' (rewrites entire aggregated result table; unsupported on Delta without full table overwrite), and 'Update' (emits only updated/new rows).
Combining 'Trigger.AvailableNow()' with Delta Lake provides the idempotence and exactly-once guarantees of streaming pipelines with the cost predictability of scheduled batch workloads.
Last updated: August 2026
7.5 Streaming Watermarking, Triggers, & Output Modes
DP-750 Exam Focus: Master stateful stream processing semantics in Spark Structured Streaming. Understand how Watermarks ( withWatermark() ) handle late-arriving event-time data and bound state store memory, configure Stateful Stream-Stream Joins , evaluate Streaming Triggers (especially Trigger.AvailableNow() ), and select appropriate Output Modes ( Append , Complete , Update ) based on aggregation and sink requirements.
1. Watermarking Mechanics & Late Data Handling
In real-time distributed data pipelines, network latency, device disconnections, and clock skews cause events to arrive out of order. If a streaming query performs stateful operations (such as tumbling window aggregations or stream-stream joins), Spark must buffer historical records in state memory (RocksDB) to ensure accurate calculations.
Without a mechanism to discard old records, the internal state store would grow infinitely over time, eventually causing worker memory exhaustion.
Watermarking defines an event-time threshold that instructs the streaming engine how long to wait for late-arriving data before finalizing window results and dropping older records from state memory. +-----------------------------------------------------------------------------------+
- | WATERMARK TIMELINE ENGINE |
- +-----------------------------------------------------------------------------------+
- | |
- | Event Time -> 12:00 12:10 12:20 12:30 12:40 12:50 |
- | | | | | | | |
- | +----------+----------+----------+----------+----------+ |
- | | |
- | Max Event Time Seen = 12:45 |
- | | |
- | [ Watermark Delay: 15 Minutes ] |
- | v |
- | Watermark Horizon = (12:45 - 15 min) ====================> 12:30 |
- | |
- | [ Arriving Event: 12:35 ] ---> ACCEPTED (Event Time >= Watermark) |
- | [ Arriving Event: 12:25 ] ---> DROPPED (Event Time < Watermark: Too Late) |
- +-----------------------------------------------------------------------------------+
- The withWatermark() API
To define a watermark, call withWatermark() on a streaming DataFrame, specifying:
timeColumn : A timestamp column representing the event time generated by the source producer.
delayThreshold : The time duration string (e.g., "10 minutes" , "2 hours" ) representing the maximum acceptable delay. Watermark Threshold = max ( EventTime seen by engine ) − DelayThreshold \text{Watermark Threshold} = \max(\text{EventTime seen by engine}) - \text{DelayThreshold} Watermark Threshold = max ( EventTime seen by engine ) − DelayThreshold from pyspark.sql.functions import col, window, count, avg
- # Tumbling 10-minute window aggregation with a 15-minute watermark
- df_windowed_metrics = (df_parsed
- .withWatermark("reading_time", "15 minutes")
- .groupBy(
- window(col("reading_time"), "10 minutes"),
- col("device_id")
- )
- .agg(
- count("*").alias("total_readings"),
- avg("temperature").alias("avg_temperature")
- ))
- How Watermarking Operates
In-Memory Retention: Any record with an event time greater than or equal to the current watermark is evaluated and updates intermediate window state in RocksDB.
Late Data Rejection: Any record with an event time strictly less than the current watermark is classified as "too late" and is dropped immediately without updating state.
State Store Eviction: Once the watermark moves past the end time of a time window (e.g., window 12:00 - 12:10 when the watermark reaches 12:11 ), the intermediate state for that window is purged from RocksDB, freeing cluster memory.
2. Stateful Stream-Stream Joins
Joining two independent real-time data streams (e.g., joining an impressions stream with a clicks stream) is a complex stateful operation. Because matching events from both streams arrive asynchronously with unpredictable delays, Spark must buffer unmatched records from both streams in the state store.
To prevent state memory from expanding indefinitely, both input streams must define watermarks and a time-range join condition . STATEFUL STREAM-STREAM JOIN MATRIX
- Stream A: Impressions (Watermark: 2 Hours)
- Stream B: Clicks (Watermark: 3 Hours)
Join Condition:
- impressions.ad_id = clicks.ad_id AND
- click_time BETWEEN impression_time AND impression_time + INTERVAL 1 HOUR
- +-------------------------------------------------------------------+
- | State Store Buffer (RocksDB) |
- | - Retains impression records for up to 1 hour waiting for click |
- | - Evicts impression once click_time window passes watermark |
+-------------------------------------------------------------------+ # Stateful Stream-Stream Inner Join with Time Bounds and Watermarks
- # 1. Define watermark on Impressions stream
- impressions = (df_impressions
- .withWatermark("impression_time", "2 hours"))
- # 2. Define watermark on Clicks stream
- clicks = (df_clicks
- .withWatermark("click_time", "3 hours"))
- # 3. Join with equality key AND event-time interval condition
- joined_stream = impressions.join(
- clicks,
- expr("""
- impressions.ad_id = clicks.ad_id AND
- click_time >= impression_time AND
- click_time <= impression_time + interval 1 hour
- """),
- joinType="inner" # Inner, left_outer, right_outer supported
- )
Exam Rule for Stream-Stream Joins: For outer joins (e.g., left_outer ), watermarks and time-range join constraints are mandatory . Spark can only emit an unmatched left row (with nulls for the right) once the right watermark has advanced past the join time boundary.
3. Streaming Triggers: Continuous, Micro-Batch, & AvailableNow
The Trigger defines the timing of streaming computation—dictating whether Spark processes records in continuous sub-millisecond loops, periodic micro-batches, or one-time incremental batch sweeps.
Trigger Type Code Syntax Operational Behavior Target Workload
Unspecified (Default) .trigger() (omitted) Executes micro-batches as fast as possible; starts next batch immediately after previous completes. Standard 24/7 low-latency streaming
Fixed Interval .trigger(processingTime='1 minute') Initiates micro-batches at strict recurring wall-clock intervals (e.g., every 60s). Predictable periodic micro-batching
AvailableNow .trigger(availableNow=True) Ingests all outstanding data across multiple micro-batches (respecting rate limits) and terminates cleanly. Scheduled incremental batch / Cost optimization
Once (Legacy) .trigger(once=True) Ingests all available data in a single micro-batch and terminates. Replaced by AvailableNow . Deprecated (risk of OOM on large backlog)
Continuous .trigger(continuous='1 second') Executes continuous low-latency processing with sub-millisecond latency (asynchronous checkpointing). Specialized low-latency (<5ms) pipelines
The Power of Trigger.AvailableNow for DP-750
Trigger.AvailableNow (introduced in Apache Spark 3.3 / DBR 10.4 LTS) is the foundational trigger for cost-effective Lakehouse architecture :
Batch Cost, Streaming Semantics: Allows data engineers to orchestrate streaming pipelines on a schedule (e.g., once every hour or night via Databricks Jobs), running on on-demand ephemeral compute clusters.
Rate-Limiting Bounded Execution: Unlike legacy Trigger.Once (which pulled all backlogged data into one monolithic batch, frequently crashing worker nodes with OOM errors), Trigger.AvailableNow honors maxFilesPerTrigger or maxOffsetsPerTrigger , splitting backlogs into safe, manageable micro-batches before stopping.
4. Structured Streaming Output Modes
The Output Mode defines what data is written to the downstream sink (e.g., Delta table, Kafka topic, or console) during each micro-batch. +-----------------------------------------------------------------------------------+
- | OUTPUT MODES SEMANTIC BEHAVIOR |
- +-----------------------------------------------------------------------------------+
- | |
- | [ OutputMode("append") ] |
- | - Default mode. |
- | - Only newly finalized rows are emitted to the sink. |
- | - With watermarked window aggregations: emits row ONLY when window closes. |
- | - Mandatory for file/Delta sinks in stateless streaming. |
- | |
- | [ OutputMode("complete") ] |
- | - Entire updated result table is rewritten to sink on every micro-batch. |
- | - Requires an aggregation query. |
- | - NOT supported on standard Delta file sinks without full overwrite. |
- | |
- | [ OutputMode("update") ] |
- | - Only rows that were modified or newly added in the current batch are emitted. |
- | - If query has no aggregations, behaves identically to 'append'. |
- | |
- +-----------------------------------------------------------------------------------+
- Detailed Output Mode Compatibility Matrix
Output Mode Stateless Queries (e.g., Filter / Select) Watermarked Aggregations Unwatermarked Aggregations Delta Lake Sink Compatible?
Append Supported (Emits all new rows) Supported (Emits rows only after window exceeds watermark) Not Supported (Cannot guarantee row will not change) Yes (Standard Bronze/Silver ingestion)
Complete Not Supported (Requires aggregation) Supported (Emits full table snapshot on every batch) Supported (Emits full table snapshot on every batch) No (Invalid for streaming append; requires custom foreachBatch or memory sink)
Update Supported (Emits new rows) Supported (Emits changed intermediate aggregation rows) Supported (Emits changed intermediate aggregation rows) Requires foreachBatch / MERGE
Exam Trap: Attempting to write a streaming aggregation query in Append mode without defining a watermark throws an AnalysisException: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark . Spark cannot know when an aggregation window is final unless a watermark is declared.