Streaming trigger modes and input-rate control | Data Engineering

Q: How does ProcessingTime work?

ProcessingTime runs a batch every N seconds of wallclock time. If a batch finishes early, Spark waits until the interval elapses. If a batch overruns the interval, the next starts immediately after. This is the standard production trigger for steady, predictable streaming. // ProcessingTime: batch e

Q: How does Once differ from AvailableNow?

Once runs a single microbatch over all available data and stops, but it ignores maxOffsetsPerTrigger, so a large backlog becomes one giant batch that can blow up memory. AvailableNow is the modern replacement: it drains all available data across multiple ratelimited batches, then stops. // Once: ONE

Q: What is Continuous mode?

Continuous mode targets millisecond endtoend latency by running longlived tasks instead of microbatches, checkpointing at the given interval. It supports only simple maplike operations, not aggregations or joins, so it is rarely used in production. // Continuous mode: millisecond latency, limited op

Q: How do you control input rate?

Input rate is controlled separately on the source with maxOffsetsPerTrigger for Kafka and maxFilesPerTrigger for file sources. These prevent a single batch from ingesting a massive backlog. // Kafka: cap at 100K records per trigger Dataset<Row kafka = spark.readStream() .format("kafka") .option("sub

Q: What are common mistakes?

The most common mistake is using Trigger.Once() with a large backlog and expecting the rate limit to help. Once ignores maxOffsetsPerTrigger, so the entire backlog becomes one massive batch. Use Trigger.AvailableNow() instead. // WRONG Once with large backlog transactions.writeStream() .trigger(Trig

Definition

ProcessingTime runs batches at a fixed interval, Once runs one batch then stops, AvailableNow drains all data in rate-limited batches then stops, Continuous is experimental low-latency, and maxOffsetsPerTrigger or maxFilesPerTrigger cap how much each batch ingests.

ProcessingTime – Trigger.ProcessingTime("30 seconds") runs a micro-batch at a fixed wall-clock interval; the standard production trigger for steady live streaming.
Once – Trigger.Once() runs exactly one batch over all available data then stops; suited for scheduled jobs but ignores rate limits, risking a giant batch.
AvailableNow – Trigger.AvailableNow() processes all available data across multiple rate-limited batches then stops; the modern, safe replacement for Once.
Continuous – Trigger.Continuous("1 second") is experimental true continuous mode for millisecond latency, supporting only simple operations.
maxOffsetsPerTrigger – Caps total Kafka records read per trigger, the primary back-pressure knob.
maxFilesPerTrigger – Caps new files picked up per trigger for file sources.

At the National Bank of India the trigger is the schedule on which the clerk processes the incoming transaction tray. Fixed-interval is "process the tray every thirty seconds." Once is "process whatever is in the tray right now, then go home" for a nightly run. AvailableNow is "clear the whole tray in as many passes as needed, respecting how many slips you may handle per pass, then go home." Continuous is "never stop, process each slip the instant it lands." The rate controls are the rule "handle at most ten thousand slips per pass."

Understanding the Concepts

What are the trigger modes?

A trigger in Structured Streaming defines when and how often a query runs a micro-batch. You set it with writeStream.trigger(...). The default trigger, when you specify none, runs micro-batches back to back: as soon as one finishes, the next starts if data is available.

JAVA

import org.apache.spark.sql.streaming.Trigger;

// Default: back-to-back micro-batches
counts.writeStream().start();

// Fixed interval: every 30 seconds
counts.writeStream()
    .trigger(Trigger.ProcessingTime("30 seconds"))
    .start();

// Once: single batch, then stop
counts.writeStream()
    .trigger(Trigger.Once())
    .start();

// AvailableNow: drain all data, then stop
counts.writeStream()
    .trigger(Trigger.AvailableNow())
    .start();

// Continuous: experimental low-latency
counts.writeStream()
    .trigger(Trigger.Continuous("1 second"))
    .start();

How does ProcessingTime work?

ProcessingTime runs a batch every N seconds of wall-clock time. If a batch finishes early, Spark waits until the interval elapses. If a batch overruns the interval, the next starts immediately after. This is the standard production trigger for steady, predictable streaming.

JAVA

// ProcessingTime: batch every 30 seconds
Dataset<Row> counts = transactions
    .groupBy("accountId")
    .count();

counts.writeStream()
    .trigger(Trigger.ProcessingTime("30 seconds"))
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "s3a://nbi-checkpoints/counts")
    .start();

DIAGRAM

Batch 0: started at 10:00:00, finished at 10:00:12 → waits until 10:00:30
Batch 1: started at 10:00:30, finished at 10:00:45 → waits until 10:01:00
Batch 2: started at 10:01:00, finished at 10:01:35 → overran, next starts immediately

How does Once differ from AvailableNow?

Once runs a single micro-batch over all available data and stops, but it ignores maxOffsetsPerTrigger, so a large backlog becomes one giant batch that can blow up memory. AvailableNow is the modern replacement: it drains all available data across multiple rate-limited batches, then stops.

JAVA

// Once: ONE giant batch, ignores rate limits
transactions.writeStream()
    .trigger(Trigger.Once())
    .option("maxOffsetsPerTrigger", 50000) // IGNORED by Once!
    .start();
// 10 million record backlog → one batch of 10M records → OOM risk

// AvailableNow: multiple bounded batches, respects rate limits
transactions.writeStream()
    .trigger(Trigger.AvailableNow())
    .option("maxOffsetsPerTrigger", 50000) // RESPECTED
    .start();
// 10M backlog → 200 batches of 50K each → safe, bounded

What is Continuous mode?

Continuous mode targets millisecond end-to-end latency by running long-lived tasks instead of micro-batches, checkpointing at the given interval. It supports only simple map-like operations, not aggregations or joins, so it is rarely used in production.

JAVA

// Continuous mode: millisecond latency, limited operations
Dataset<Row> filtered = transactions
    .filter(col("amount").gt(new java.math.BigDecimal("100000.00")));

filtered.writeStream()
    .trigger(Trigger.Continuous("1 second"))  // checkpoint every 1s
    .outputMode("append")
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("topic", "nbi-alerts")
    .start();

// NOT supported in Continuous mode:
// transactions.groupBy("accountId").count()  // AnalysisException

How do you control input rate?

Input rate is controlled separately on the source with maxOffsetsPerTrigger for Kafka and maxFilesPerTrigger for file sources. These prevent a single batch from ingesting a massive backlog.

JAVA

// Kafka: cap at 100K records per trigger
Dataset<Row> kafka = spark.readStream()
    .format("kafka")
    .option("subscribe", "nbi-transactions")
    .option("maxOffsetsPerTrigger", 100000)
    .load();

// File source: cap at 10 files per trigger
Dataset<Row> files = spark.readStream()
    .format("parquet")
    .option("maxFilesPerTrigger", 10)
    .schema(txnSchema)
    .load("s3a://nbi-data/incoming/");

How do triggers and rate limits work together?

Triggers and rate limits work together to create different operational patterns. ProcessingTime plus a rate limit keeps live processing steady. AvailableNow plus a rate limit drains a backlog safely and stops.

| Pattern | Trigger | Rate Limit | Use Case |
|---|---|---|---|
| Steady live processing | ProcessingTime(30s) | 50K per trigger | Real-time dashboard |
| Cost-saving nightly drain | AvailableNow | 50K per trigger | Batch-like scheduled job |
| One-shot (legacy) | Once | Ignored | Old scheduled job |
| Ultra-low latency | Continuous(1s) | N/A | Simple alerting |

JAVA

// Pattern 1: Steady live processing
counts.writeStream()
    .trigger(Trigger.ProcessingTime("30 seconds"))
    .start();  // Runs continuously, predictable cadence

// Pattern 2: Cost-saving nightly drain
counts.writeStream()
    .trigger(Trigger.AvailableNow())
    .start();  // Drains backlog, then cluster shuts down

What are common mistakes?

The most common mistake is using Trigger.Once() with a large backlog and expecting the rate limit to help. Once ignores maxOffsetsPerTrigger, so the entire backlog becomes one massive batch. Use Trigger.AvailableNow() instead.

JAVA

// WRONG -- Once with large backlog
transactions.writeStream()
    .trigger(Trigger.Once())
    .option("maxOffsetsPerTrigger", 50000) // IGNORED!
    .start(); // Ingests entire backlog in one batch → OOM

// CORRECT -- AvailableNow with rate limit
transactions.writeStream()
    .trigger(Trigger.AvailableNow())
    .option("maxOffsetsPerTrigger", 50000) // Respected
    .start(); // Drains backlog in bounded batches

What should a senior engineer know?

The key insight is that triggers and rate limits are separate controls that compose. ProcessingTime controls cadence, rate limits control batch size, and AvailableNow adds a stop-when-done behavior. For cost-efficient scheduled streaming on a backlog, AvailableNow plus maxOffsetsPerTrigger is the standard pattern. For steady live processing, ProcessingTime plus a rate limit prevents the job from falling hopelessly behind.

Visual Overview

DIAGRAM

  Trigger Modes Comparison

  ProcessingTime(30s):
  ──▶ Batch ──▶ wait ──▶ Batch ──▶ wait ──▶ Batch ──▶ ...
      │  12s  │  18s  │  25s  │  5s   │  10s  │
      └───────┘       └───────┘       └───────┘
      Runs forever at fixed cadence

  Once():
  ──▶ ONE GIANT BATCH ──▶ STOP
      │  all data      │
      └────────────────┘
      Ignores maxOffsetsPerTrigger!

  AvailableNow():
  ──▶ Batch ──▶ Batch ──▶ Batch ──▶ Batch ──▶ STOP
      │ 50K │  │ 50K │  │ 50K │  │ 30K │
      └─────┘  └─────┘  └─────┘  └─────┘
      Respects rate limits, stops when drained

  Continuous(1s):
  ──▶ record ──▶ record ──▶ record ──▶ ...
      │  1ms  │  │  2ms  │  │  1ms  │
      No micro-batches, checkpoint every 1s
      Limited to map-like operations

Key Takeaway

A trigger in Structured Streaming controls when and how often a query runs a micro-batch. ProcessingTime with a fixed interval runs a batch on that wall-clock cadence and is the standard production trigger for steady live streaming. Once runs a single batch over all available data and stops, which was used for scheduled incremental jobs, but its weakness is that it ignores the per-trigger rate limit, so a large backlog becomes one giant batch.

AvailableNow is the modern replacement: it drains all available data across multiple rate-limited micro-batches, then stops, making it safe for large backlogs. Continuous mode is experimental, aims for millisecond latency with long-lived tasks, supports only simple operations, and is rarely used in production.

Input rate is controlled separately on the source. For Kafka, maxOffsetsPerTrigger caps the total records read per trigger. For file sources, maxFilesPerTrigger caps files per trigger. The key insight is that triggers and rate limits work together: AvailableNow plus a rate limit drains a backlog cost-efficiently, while ProcessingTime plus a rate limit keeps live processing steady.

At the National Bank of India Rajesh used ProcessingTime of thirty seconds with a 50,000 record per-trigger cap for a live fraud-scoring stream, giving predictable cadence and bounded batches. For the nightly reconciliation he used AvailableNow with the same rate limit, so it drained the entire day's backlog in several safe batches and then stopped to release the cluster.

Remember one sentence: ProcessingTime for steady live cadence, AvailableNow for cost-saving backlog draining that respects rate limits, and maxOffsetsPerTrigger as the back-pressure that keeps every batch bounded.

---