Batch vs Stream Ingestion for IoT Data

Definitions and scope

This page is for architects and data engineers deciding how sensor, meter and maintenance data should move into storage. It is about the ingestion style, not the transport: Kafka vs MQTT covers transports, and designing an IoT pipeline with Kafka covers a full streaming pipeline.

• Batch ingestion loads a bounded set of data, known in full before the job starts, as one unit: yesterday's half-hourly meter file, last night's CMMS work-order extract, a gateway's buffered week of readings after an outage. The Flink documentation describes its BATCH mode as for bounded jobs with a known fixed input that do not run continuously.
• Stream ingestion processes an unbounded input continuously, record by record or in small groups, as it arrives: BMS points published every 10 to 60 seconds, alarms, occupancy events. Flink's STREAMING mode is for unbounded jobs that need continuous incremental processing and stay online indefinitely.
• Micro-batch sits between them: a streaming job that processes the stream as a series of small batch jobs. The Spark documentation says this is how Structured Streaming runs queries by default.

Keep the layers separate in your head. A streaming platform (Kafka) or broker (MQTT) moves events; a database stores and serves them; a warehouse or lakehouse holds history for analytics. Batch or stream describes how data crosses each boundary, and a single estate usually uses both.

Which IoT and facilities data fits which style

The data source usually decides. A realistic mix for a facilities operator running a BMS, sub-metering and a CMMS:

Trade-offs

Batch loading in PostgreSQL

For a bounded file the PostgreSQL documentation is clear: use COPY to load all rows in one command; it has significantly less overhead than a series of INSERTs. Load into a staging table first, validate, then move good rows into the readings table with an idempotent INSERT ... SELECT ... ON CONFLICT so that re-running the job for the same file is harmless.

The current PostgreSQL documentation (18) lists ON_ERROR and REJECT_LIMIT options for COPY; check that your server version supports them before relying on them. Note the scope: ON_ERROR ignore skips rows whose values cannot be converted to the column type. It does not skip rows that violate a CHECK constraint, which still fail the command, because COPY FROM invokes check constraints on the destination table. Put range and plausibility checks in the validation step, as described in IoT data quality checks.

-- server-side file; from a client use psql's \copy instead
COPY meter_staging (meter_id, interval_start, kwh)
FROM '/data/incoming/meters_2026-10-26.csv'
WITH (FORMAT csv, HEADER true, ON_ERROR ignore, REJECT_LIMIT 50);

Incremental batch jobs: do not trust an event-time high-water mark

Many batch jobs run every few minutes or hours over "what is new since last time": refreshing roll-ups, exporting to the warehouse, feeding a CMMS rule engine. The tempting high-water mark is the largest event timestamp processed so far. It fails with IoT data, because a gateway that reconnects delivers old timestamps after newer ones have already been processed. Those late readings sit below the mark and are never picked up.

Recommended for roll-ups: recompute a bounded recent window on every run. Instead of a high-water mark, each run recomputes every hour bucket inside a recent window (for example the last 3 hours) from the raw rows committed at that moment, with an idempotent upsert, and a slower reconciliation run does the same for older closed periods. Its correctness rests only on documented PostgreSQL behaviour (READ COMMITTED snapshots and the atomic outcome of INSERT ... ON CONFLICT DO UPDATE), as set out in continuous aggregates vs materialized views. A late reading is therefore included by the first run after its commit, provided its hour is still inside the window. The script below was run in a single session on PostgreSQL 18.6 (PGlite); concurrent sessions cannot be run here, so the run times are simulated and the 3-hour window is an illustrative assumption.

SET TIME ZONE 'UTC';
CREATE TABLE reading (
    point_id   integer NOT NULL,
    ts         timestamptz NOT NULL,
    value      double precision NOT NULL,
    ingest_id  bigint GENERATED ALWAYS AS IDENTITY,   -- not used by this job
    PRIMARY KEY (point_id, ts)
);
CREATE TABLE reading_hourly (
    point_id   integer NOT NULL,
    hour       timestamptz NOT NULL,
    avg_value  double precision NOT NULL,
    samples    integer NOT NULL,
    PRIMARY KEY (point_id, hour)
);
INSERT INTO reading (point_id, ts, value)
SELECT 101, g, 13 + extract(minute FROM g) / 60.0
FROM generate_series(timestamptz '2026-10-27 08:00+00', '2026-10-27 09:59+00',
                     interval '1 minute') AS g;

-- Recompute every hour from p_from (rounded down) to p_to from committed rows
CREATE PROCEDURE recompute_reading_hourly(p_from timestamptz, p_to timestamptz DEFAULT 'infinity')
LANGUAGE sql AS $$
  INSERT INTO reading_hourly (point_id, hour, avg_value, samples)
  SELECT point_id, date_trunc('hour', ts), avg(value), count(*)
  FROM reading
  WHERE ts >= date_trunc('hour', p_from) AND ts < p_to
  GROUP BY 1, 2
  ON CONFLICT (point_id, hour) DO UPDATE
  SET avg_value = EXCLUDED.avg_value, samples = EXCLUDED.samples;
  -- hours in the range whose raw rows have all been deleted
  DELETE FROM reading_hourly h
  WHERE h.hour >= date_trunc('hour', p_from) AND h.hour < p_to
    AND NOT EXISTS (SELECT 1 FROM reading r
                    WHERE r.point_id = h.point_id
                      AND r.ts >= h.hour AND r.ts < h.hour + interval '1 hour');
$$;

-- Run at 10:00 with an illustrative 3-hour window (in a real job: now() - interval '3 hours')
CALL recompute_reading_hourly(timestamptz '2026-10-27 10:00+00' - interval '3 hours');
SELECT point_id, hour, round(avg_value::numeric, 3) AS avg_value, samples FROM reading_hourly ORDER BY hour;

-- The late reading stamped 08:00:30 commits after that run
INSERT INTO reading (point_id, ts, value) VALUES (101, '2026-10-27 08:00:30+00', 13.9);

-- Next run at 10:10: the 08:00 hour is still inside the window
CALL recompute_reading_hourly(timestamptz '2026-10-27 10:10+00' - interval '3 hours');
SELECT point_id, hour, round(avg_value::numeric, 3) AS avg_value, samples FROM reading_hourly ORDER BY hour;

-- after the 10:00 run
 point_id |          hour          | avg_value | samples
----------+------------------------+-----------+---------
      101 | 2026-10-27 08:00:00+00 |    13.492 |      60
      101 | 2026-10-27 09:00:00+00 |    13.492 |      60

-- after the 10:10 run: the late reading is included
 point_id |          hour          | avg_value | samples
----------+------------------------+-----------+---------
      101 | 2026-10-27 08:00:00+00 |    13.498 |      61
      101 | 2026-10-27 09:00:00+00 |    13.492 |      60

Limits of the window recompute

• Window length. The window must cover the longest expected gap between a reading's event time and its commit (gateway buffering plus the longest writer transaction, because a row is visible only at commit) plus the interval between runs. Readings later than that are left to the reconciliation run, so schedule one over recently closed periods (for example the previous day) and keep its range clear of raw data removed by retention, because the DELETE step removes roll-up rows whose raw rows are gone.
• Cost. Each run re-reads the whole window, so cost grows with window length times data rate rather than with the number of new rows.
• Overlapping runs do not break correctness, but waste work, and the run that commits last writes its own snapshot's figures until the next run; let the scheduler run one at a time.
• Exports rather than roll-ups. For a warehouse export the same idea applies: re-export the recent window each run and make the target load idempotent (an upsert or MERGE on the reading key), instead of exporting only rows above a watermark.

Educational example: an ingest-sequence watermark

Educational example: shows the idea in a single session; not safe for production as shown, because of the race described below. This pattern records an ingest sequence on every row (an identity column), remembers the last sequence processed, and re-aggregates every time bucket touched by new rows, late or not. It is cheaper per run than the window recompute and shows clearly why an event-time watermark misses late rows. It was run in a single session, which is the case where an identity column reflects arrival order; the caveat after it explains what goes wrong with several concurrent writers.

CREATE TABLE reading (
    point_id   integer NOT NULL,
    ts         timestamptz NOT NULL,
    value      double precision NOT NULL,
    ingest_id  bigint GENERATED ALWAYS AS IDENTITY,
    PRIMARY KEY (point_id, ts)
);
CREATE TABLE reading_hourly (
    point_id   integer NOT NULL,
    hour       timestamptz NOT NULL,
    avg_value  double precision NOT NULL,
    samples    integer NOT NULL,
    PRIMARY KEY (point_id, hour)
);
CREATE TABLE job_watermark (
    job_name       text PRIMARY KEY,
    last_ingest_id bigint NOT NULL
);
INSERT INTO job_watermark VALUES ('hourly_rollup', 0);

INSERT INTO reading (point_id, ts, value)
SELECT 101, g, 13 + extract(minute FROM g) / 60.0
FROM generate_series(timestamptz '2026-10-27 08:00+00', '2026-10-27 09:59+00',
                     interval '1 minute') AS g;

BEGIN;
WITH bounds AS (
    SELECT w.last_ingest_id AS lo, (SELECT max(ingest_id) FROM reading) AS hi
    FROM job_watermark w WHERE w.job_name = 'hourly_rollup' FOR UPDATE
), touched AS (
    SELECT DISTINCT r.point_id, date_trunc('hour', r.ts) AS hour
    FROM reading r, bounds b
    WHERE r.ingest_id > b.lo AND r.ingest_id <= b.hi
), agg AS (
    INSERT INTO reading_hourly (point_id, hour, avg_value, samples)
    SELECT t.point_id, t.hour, avg(r.value), count(*)
    FROM touched t
    JOIN reading r ON r.point_id = t.point_id
                  AND r.ts >= t.hour AND r.ts < t.hour + interval '1 hour'
    GROUP BY t.point_id, t.hour
    ON CONFLICT (point_id, hour) DO UPDATE
        SET avg_value = EXCLUDED.avg_value, samples = EXCLUDED.samples
    RETURNING 1
)
UPDATE job_watermark SET last_ingest_id = (SELECT hi FROM bounds)
WHERE job_name = 'hourly_rollup'
RETURNING last_ingest_id, (SELECT count(*) FROM agg) AS buckets_written;
COMMIT;

 last_ingest_id | buckets_written
----------------+-----------------
            120 |               2

 point_id |          hour          | avg_value | samples
----------+------------------------+-----------+---------
      101 | 2026-10-27 08:00:00+00 |    13.492 |      60
      101 | 2026-10-27 09:00:00+00 |    13.492 |      60

A late reading arrives (educational example continued)

A gateway now delivers a reading stamped 08:00:30, an hour and a half late. An event-time watermark (the last processed ts was 09:59) finds nothing new; the ingest sequence finds it. Running the same job again re-aggregates only the 08:00 hour.

INSERT INTO reading (point_id, ts, value) VALUES (101, '2026-10-27 08:00:30+00', 13.9);

SELECT count(*) AS missed_by_event_time_watermark
FROM reading WHERE ts > timestamptz '2026-10-27 09:59+00';
--  missed_by_event_time_watermark
-- --------------------------------
--                               0

SELECT count(*) AS found_by_ingest_watermark
FROM reading
WHERE ingest_id > (SELECT last_ingest_id FROM job_watermark WHERE job_name = 'hourly_rollup');
--  found_by_ingest_watermark
-- ---------------------------
--                          1

-- run the batch job again (same statement as above):
--  last_ingest_id | buckets_written
-- ----------------+-----------------
--             121 |               1

--  point_id |          hour          | avg_value | samples
-- ----------+------------------------+-----------+---------
--       101 | 2026-10-27 08:00:00+00 |    13.498 |      61
--       101 | 2026-10-27 09:00:00+00 |    13.492 |      60

Caveat: concurrent writers and commit order (not verified here)

The job above was run in one session; behaviour with concurrent writers was not verified here, and the following is an analysis from the PostgreSQL documentation. An identity column draws its values from a sequence. nextval hands each session a distinct value at insert time, but the value is not reclaimed if the transaction aborts, and with sequence caching values can be generated out of order across sessions. Nothing ties the order of the values to the order in which transactions commit. So, with two writers, transaction A can take ingest_id 121, transaction B can take 122 and commit first, the job can read a maximum of 122 and move the watermark there, and when A commits its row at 121 is already below the watermark and is never processed. Gaps left by aborted transactions are harmless to this job; rows that commit late are the problem.

Documented options, not verified here. These are listed for readers who want to keep a watermark design; the recommended path for roll-ups remains the window recompute above, whose correctness does not depend on locking or timing that could not be exercised here.

• Lock out writers while reading the maximum. Take LOCK TABLE reading IN SHARE MODE at the start of the job transaction. SHARE conflicts with the ROW EXCLUSIVE lock every INSERT holds, so the job waits until in-flight writers have committed or rolled back, and new inserts wait until the job commits; every ingest_id up to the maximum it then reads is final. The cost is that ingestion pauses for the length of the job.
• A time margin. Only process rows older than a margin, using an ingested_at timestamptz DEFAULT now() column. This is only safe if no writer transaction can stay open longer than the margin, because now() is the transaction start time; on PostgreSQL 17 and later, transaction_timeout can enforce that limit for the writer role. The margin itself is a judgement based on your longest writer transaction.
• One writer per table, or drive the job from the stream itself (Kafka offsets) rather than from the table.

Without one of them the watermark example is only correct with a single writer, and none of them was exercised with concurrent writers here.

BEGIN;
LOCK TABLE reading IN SHARE MODE;   -- waits for writers in flight, blocks new inserts until COMMIT
-- ... the same WITH bounds ... UPDATE job_watermark statement as above ...
COMMIT;

Stream ingestion into a database

Streaming into a relational database rarely means one transaction per message. The KafkaConsumer javadoc's own database example buffers records and inserts them as a batch before committing the offset, and the PostgreSQL documentation explains why: if each insertion is committed separately, the server does a lot of work for every row. A writer that collects readings for a short interval or up to a fixed number of records before each commit trades a little latency for less per-commit work; the right interval and batch size depend on your hardware and latency target, so measure them rather than copying a figure.

Two consequences follow. First, delivery is at least once, so writes must be idempotent; see Kafka consumer idempotency for the offset-and-upsert pattern. Second, stream processors that compute results before storage (windows, joins, enrichment) need state and checkpoints. The Spark documentation describes its default micro-batch engine as reaching end-to-end latencies as low as 100 milliseconds with exactly-once fault tolerance; that is the project's own statement about Spark's processing, not a guarantee that each reading is written exactly once to your database (that depends on the sink: without offsets stored atomically with the data and a fencing check, a database sink is at-least-once with idempotent writes; see fencing an outdated consumer), and actual latency depends on your sources, sinks and cluster.

Flink makes the boundedness point explicit: the same DataStream program can run in STREAMING or BATCH mode, and over a bounded input it produces the same final result either way, though streaming emits incremental updates (the documentation compares them to upserts in a database) while batch emits one final result. That is a useful mental model for IoT too: the stream keeps the table current; a batch over a closed window settles the final numbers.

Decision guide and when not to stream

• Stream when someone or something acts on the data within minutes: live plant dashboards, alarm and fault-detection rules, demand-response signals, occupancy-driven control.
• Batch when the input is naturally bounded (supplier files, daily extracts), when the consumer is a report or a warehouse, or when the data must be validated as a whole before anyone sees it, such as billing-grade meter data.
• Both is, in our editorial view, the common answer for a BMS estate: stream raw readings into the operational store (see how to store IoT sensor data), then batch roll-ups, energy reports and warehouse exports from it.

When not to stream. Our editorial view: a site with a few hundred points polled every 15 minutes rarely needs Kafka and a stream processor; a scheduled job that polls and inserts in one transaction is easier to run and to recover. Streaming also adds always-on services to patch and monitor. Adopt it when freshness requirements, data volume or the number of consumers justify the operational cost, not because the data is "real-time" by nature.

Limitations of this page. It compares styles qualitatively. Throughput and latency depend on hardware, schema, batch sizes and network, so measure on your own workload before you commit to a design; the time-series overview explains how the storage side fits in.

Frequently asked questions

Sources

• Apache Flink documentation: Execution Mode (Batch/Streaming)
• Apache Spark documentation: Structured Streaming Programming Guide
• Apache Kafka: Introduction
• Apache Kafka 4.4 javadoc: KafkaConsumer
• PostgreSQL 18 documentation: Populating a Database
• PostgreSQL 18 documentation: COPY
• PostgreSQL 18 documentation: Sequence Manipulation Functions
• PostgreSQL 18 documentation: CREATE SEQUENCE
• PostgreSQL 18 documentation: Explicit Locking
• PostgreSQL 18 documentation: Date/Time Functions
• PostgreSQL 18 documentation: Client Connection Defaults
• PostgreSQL 18 documentation: Transaction Isolation
• PostgreSQL 18 documentation: INSERT (ON CONFLICT)

Checked 27 October 2026.

How we research these articles: our editorial method. Research-based from official documentation; SQL examples are run on a local database where the page says so. CodeWithSQL earns nothing from the vendors mentioned.