Hi @YuriS,

There are a few things going on here, and I will walk through each one.

INPUTROWSPERSECOND SHOWING 0

The inputRowsPerSecond metric is not calculated from the current batch. It is the rate of data arriving between the end of the previous trigger and the start of the current one. Specifically, Spark computes it as:

inputRowsPerSecond = numInputRows / (elapsed time since previous trigger completed)

On the very first batch after a stream starts (or restarts), there is no previous trigger to measure against, so the elapsed time is either zero or undefined. Spark reports 0.0 in that case. You will also see 0.0 whenever the time gap between triggers is extremely small or when the trigger spent most of its duration on metadata operations (latestOffset, WAL commit, query planning) rather than actively fetching rows.

If you need a reliable throughput metric, processedRowsPerSecond is generally more useful because it divides numInputRows by the actual triggerExecution duration of the current batch:

processedRowsPerSecond = numInputRows / durationMs.triggerExecution

This gives you the actual processing rate for the batch that just completed.

NUMINPUTROWS BEING 3-5X LARGER THAN ACTUAL COMMITTED ROWS

This is the more interesting part of your observation. There are a few possible explanations, and they can stack:

1. SPECULATIVE EXECUTION AND TASK RETRIES
If spark.speculation is enabled (it is by default on many Databricks cluster configurations), Spark may launch duplicate tasks for the same partition. Each speculative copy reads the same input rows. numInputRows counts all rows read across all task attempts, not just the ones from the winning task. If you have 3 speculative copies that each read the same data, numInputRows can report 3x the actual unique rows.

Check whether speculation is enabled:

spark.conf.get("spark.speculation")

If it is "true" and you want numInputRows to reflect unique rows, you can disable it:

spark.conf.set("spark.speculation", "false")

However, weigh this against the benefit speculation gives you for straggler mitigation.

2. TASK FAILURES AND RETRIES
If any tasks fail and are retried (even without speculation), the retried tasks re-read the same input rows. numInputRows accumulates across all attempts.

3. FOREACHBATCH RETRY SEMANTICS
When using foreachBatch, if the batch function throws an exception partway through, the entire micro-batch may be retried. The retried execution re-reads all the input rows from the source, and numInputRows will include both the failed and successful reads.

4. DELTA SOURCE INTERNAL READS
When reading from a Delta table as a streaming source with Change Data Feed or certain merge/update patterns upstream, the Delta source may emit multiple row versions (insert, update_preimage, update_postimage, delete) for a single logical row change. These all count toward numInputRows even if your foreachBatch logic filters some of them out before writing to the target table.

HOW TO RECONCILE THE NUMBERS

For accurate row counts of what actually landed in your target, the approach you are already using (comparing batch_id in the target table) is the correct one. The StreamingQueryListener metrics are operational telemetry for throughput estimation, not precise accounting.

If you need exact input counts inside your foreachBatch logic, use the DataFrame.observe() API to count rows at the point in the pipeline you care about:

from pyspark.sql.functions import count, lit

def process_batch(df, batch_id):
  observed = df.observe("row_count", count(lit(1)).alias("cnt"))
  # your processing logic on observed
  ...

query = (
  spark.readStream
      .format("delta")
      .table("source_table")
      .writeStream
      .foreachBatch(process_batch)
      .option("checkpointLocation", "/path/to/checkpoint")
      .trigger(processingTime="1 minute")
      .start()
)

You can then capture the observed metric in your StreamingQueryListener.onQueryProgress handler via event.progress.observedMetrics.

BATCH VS. TRIGGER

A trigger is the scheduling mechanism that tells Spark when to look for new data (e.g., processingTime="1 minute" fires every 60 seconds). A batch (micro-batch) is the unit of work that actually processes the data found during that trigger.

In practice, they are almost always 1:1: one trigger fires, one micro-batch executes. The distinction surfaces in a few cases:

- If a trigger fires but no new data is available, no new batch is created (batchId does not increment), and an onQueryIdle event fires instead of onQueryProgress.
- After a restart, the first trigger may process a backlog, but it still runs as a single micro-batch.
- The batchDuration and durationMs.triggerExecution values will be identical for a normal micro-batch cycle, which is why they look the same in your metrics.

The key metrics reference for all of these fields is here:
https://learn.microsoft.com/en-us/azure/databricks/structured-streaming/stream-monitoring

SUMMARY

- inputRowsPerSecond = 0 is expected on the first batch and when metadata time dominates. Use processedRowsPerSecond for throughput monitoring.
- numInputRows includes retried and speculative task reads. It is not a count of unique input rows. Check spark.speculation and task retry counts in the Spark UI.
- For precise row accounting, use DataFrame.observe() or compare offsets/target table data as you are already doing.
- Batch and trigger are conceptually different but 1:1 in normal micro-batch operation.

REFERENCES

- Monitoring Structured Streaming queries:

https://learn.microsoft.com/en-us/azure/databricks/structured-streaming/stream-monitoring

- StreamingQueryListener documentation:

https://docs.databricks.com/en/structured-streaming/stream-monitoring.html

- Observable metrics with DataFrame.observe():

https://docs.databricks.com/en/structured-streaming/stream-monitoring.html#defining-observable-metrics-in-structured-streaming

* This reply used an agent system I built to research and draft this response based on the wide set of documentation I have available and previous memory. I personally review the draft for any obvious issues and for monitoring system reliability and update it when I detect any drift, but there is still a small chance that something is inaccurate, especially if you are experimenting with brand new features.

If this answer resolves your question, could you mark it as "Accept as Solution"? That helps other users quickly find the correct fix.