Batch and Incremental Processing

Batch processing handles a finite group of records during one execution. A batch might contain records created during the last hour, a fixed number of rows, or everything created since the previous successful run.

The goal is not to read the entire source on every execution. A well-designed job can start with a full load, then process only new data while keeping its processing boundary explicit.

What is batch processing?#

The basic flow is:

Source   ↓select batch   ↓DataFrame   ↓transform   ↓write result

Batch does not necessarily mean once per day. A job that runs every minute can still process one batch per run. The interval and the amount of data are separate design decisions.

Full load vs. incremental load#

A full load reads all available source data:

df = client.query_df("""    SELECT *    FROM raw.orders""")

This is easy to understand and useful for an initial load or a controlled rebuild. It becomes inefficient as the source grows because every run repeats work for records that have already been processed.

An incremental load asks a narrower question: what data has appeared since the last successfully processed point?

Run 1: [--------------------] process initial dataRun 2: [----]               process only new dataRun 3: [---]                process only new data

Incremental processing reduces data movement and work per run, but it requires state, clear boundaries, and a strategy for reruns and late-arriving records.

The hybrid Python + SQL approach#

This article uses SQL to select a batch in ClickHouse and Python/Pandas to transform it:

Python   ↓SQL query   ↓ClickHouse   ↓Pandas DataFrame   ↓Python transformation   ↓ClickHouse

SQL is useful for selecting only the records that belong to the current batch. Pandas is useful when the selected data needs Python-side inspection or transformation. The SQL here is supporting material, not a replacement for a dedicated SQL guide.

Processing a simple batch#

Once a query has selected the current batch, the transformation can be small and explicit:

import pandas as pd

df = client.query_df("""    SELECT        order_id,        customer_id,        created_at,        price,        quantity    FROM raw.orders    LIMIT 10000""")
if df.empty:    print("No rows in this batch")    return
df["created_at"] = pd.to_datetime(    df["created_at"],    utc=True,)
df["total"] = (    df["price"] * df["quantity"])

The batch is read, checked for emptiness, converted to a consistent datetime type, and given a derived total. The transformation is intentionally simple; the Data Transformations with Pandas article covers DataFrame transformation patterns in more depth.

Understanding time windows#

A fixed time window defines the interval represented by a batch. Conceptually:

start = window_startend   = window_end

The corresponding SQL condition is:

WHERE created_at >= {start:DateTime}  AND created_at < {end:DateTime}

The {start:DateTime} and {end:DateTime} parts are not universal SQL syntax. They are ClickHouse server-side parameter placeholders: start and end are parameter names, while DateTime declares the ClickHouse type. ClickHouse Connect receives the actual values through the parameters dictionary. A different database driver may use a different placeholder style.

The interval is half-open: [start, end). It includes the start and excludes the end.

Adjacent windows then fit together without overlap:

start <= created_at < endend   <= created_at < next_end

A record exactly at end belongs to the next window. Consistent boundaries prevent both duplicate processing and gaps between batches.

High-water mark / watermark#

A high-water mark, also called a watermark, is the latest successfully processed position. For timestamp-based data, it is a value from the source ordering field, such as created_at or a reliable ingestion timestamp.

The primary incremental mental model is:

last processed position          ↓WHERE created_at > last_processed_at          ↓new records only

Timestamps are only one possible watermark. Other choices include:

• an increasing ID
• a sequence number
• an event offset
• a source version
• an ingestion position

The best watermark is a field whose ordering and arrival guarantees are understood.

Where the watermark comes from#

For a simple pipeline, the destination can provide the latest processed value:

result = client.query("""    SELECT maxOrNull(created_at)    FROM staging.orders""")
last_ts = result.result_columns[0][0]

Read the expression step by step:

• client.query(...) sends a query to ClickHouse.
• .result_columns represents the result as columns.
• [0] selects the first result column.
• The second [0] selects the first value in that column.

Therefore, last_ts is the scalar returned by maxOrNull(). The OrNull variant is useful because an empty Staging table returns NULL, which becomes None in Python instead of requiring an artificial starting timestamp.

First run and next runs#

On the first run there is no watermark, so the job performs an initial load:

if last_ts is None:    sql = """        SELECT            order_id,            customer_id,            created_at,            price,            quantity        FROM raw.orders        ORDER BY created_at    """
    df = client.query_df(sql)

On later runs, use a parameterized query to read only records after the watermark:

sql = """    SELECT        order_id,        customer_id,        created_at,        price,        quantity    FROM raw.orders    WHERE created_at > {last_ts:DateTime}    ORDER BY created_at"""
df = client.query_df(    sql,    parameters={"last_ts": last_ts},)

The parameters dictionary binds the value separately from the SQL string. This is ClickHouse Connect's parameter-binding API, not a portable SQL feature. It is safer and clearer than interpolating an external value into the query. Use the ClickHouse type that matches the source column; for a DateTime64 column, the placeholder type should match that schema.

Empty batches#

An incremental query may return no rows. That is normal when a scheduled run has no new data:

if df.empty:    print("No new data")    return

An empty batch is not necessarily an error. Treating it as a successful no-op keeps the schedule healthy and avoids unnecessary transformations or writes.

The decision flow is:

Scheduled run      ↓New records?   /       \ No         Yes ↓           ↓finish     transformsuccessfully

Batch boundaries#

Time-window processing and watermark processing solve related but different problems. A window describes the interval a batch represents:

WHERE created_at >= {start:DateTime}  AND created_at < {end:DateTime}

A watermark describes where processing stopped:

WHERE created_at > {last_ts:DateTime}

Use half-open intervals for fixed windows. The first batch may be [start, end), and the next may be [end, next_end). A record at exactly end belongs to exactly one batch.

Watermark-based vs. fixed-window processing#

Watermarks are convenient for “everything new since the last run.” Fixed windows are useful when runs correspond to explicit periods that can be rebuilt independently.

No lookback in the basic watermark pattern#

The basic pattern uses a strict high-water mark:

last processed timestamp          ↓WHERE created_at > last_ts          ↓new records only

This is intentionally the simplest pattern. It is safe only when records do not arrive late with timestamps older than or equal to the current watermark, or when the watermark is based on a reliable ingestion or ordering field.

The full practical example later in this article uses ingested_at as that ordering field. This lets it process a late business event when the source records its arrival time reliably.

Incremental loading without a lookback is not automatically correct for every source. The source's arrival and ordering guarantees must be part of the design.

Late-arriving data#

Suppose the current watermark is last_ts, representing position 12:00 in a conceptual timeline. A record may arrive later with created_at equal to 11:58. A query using created_at > last_ts will never retrieve it because the record falls behind the already saved boundary.

The delay can be minutes, days, months, or even five years. For example, a source system may be offline for a long time, replay an old export, or send a historical correction. If that record has an old created_at value but arrives now, a strict watermark on created_at treats it as already processed and misses it.

When the source provides both business time and arrival time, keep them separate:

created_at  → when the order or event happenedingested_at  → when the pipeline received it

The pipeline can use ingested_at as its watermark while preserving created_at for reporting and business logic:

SELECT *FROM raw.ordersWHERE ingested_at > {last_ingested_at:DateTime}

This captures a five-year-old event that was ingested today. It does not remove the need for deduplication or recomputing affected aggregates: a late record may change results that were already served to consumers.

Common production responses include:

1. Use an ingestion timestamp rather than a business event timestamp.
2. Use a monotonically increasing sequence or ID.
3. Use change data capture (CDC) when the source supports it.
4. Persist explicit processing state with a defined recovery policy.
5. Re-read a small overlap, or lookback window, and deduplicate the result.

Lookback can improve recovery from late arrivals, but it adds duplicate-handling and state-management requirements. It belongs in the reliability design, not as an automatic addition to every basic query.

Duplicate timestamps#

Even without late arrivals, a timestamp may not uniquely identify a record. Several orders can share the same created_at value. If the last processed record and another unprocessed record have the same timestamp, a strict timestamp-only watermark can be ambiguous.

A composite watermark can make the position deterministic:

(created_at, order_id)

The next query then continues after the exact pair of values, using a lexicographic comparison appropriate for the database. The primary example stays timestamp-based for clarity, but uniqueness and ordering must be checked before using it in production.

Reruns and idempotency#

Consider a batch that reads 500 rows, transforms them, begins inserting, and then fails. The job may be retried or run manually again.

Batch processing should therefore be designed together with idempotency. A rerun should not create an incorrect duplicate or inconsistent result.

Common strategies include:

• deterministic record keys
• upserts
• replacing a known partition
• deduplication
• an explicit processed-state table

The implementation depends on the target table and failure model. The Data Pipeline article introduces idempotency and safe reruns in the broader pipeline architecture.

When should the watermark advance?#

The watermark must represent successfully processed data. Do not conceptually advance processing state before the destination write succeeds:

Read batch    ↓Transform    ↓Validate    ↓Write successfully    ↓Now this range is processed

The unsafe sequence is:

Read batch    ↓Mark processed    ↓Transformation fails    ↓Data is skipped forever

For a simple destination-derived watermark, a successful insert makes the destination state visible to the next run. More complex pipelines usually need explicit state and a defined commit or recovery strategy.

A dedicated state table#

Using MAX(destination.timestamp) is convenient for simple pipelines. A dedicated state table makes processing state explicit:

pipeline_state
pipeline_namelast_processed_atupdated_at

This approach can record state even when the destination does not contain a comparable timestamp, and it can support multiple independent partitions or sources. It also creates another state transition that must be updated only after successful processing.

Full practical example#

This function follows the complete sequence: connect, find the current processing boundary, read the right batch, handle an empty result, transform, select the output schema, and write the result. It uses ingested_at as the watermark so that a record with an old created_at can still be processed when it arrives later. The source and target tables must therefore preserve ingested_at.

import pandas as pdfrom clickhouse_connect import get_client

def process_new_orders():    client = get_client(        host="clickhouse",        port=8123,        username="default",        password="password",    )
    result = client.query("""        SELECT maxOrNull(ingested_at)        FROM staging.orders    """)
    last_ingested_at = result.result_columns[0][0]
    if last_ingested_at is None:        sql = """            SELECT                order_id,                customer_id,                created_at,                ingested_at,                price,                quantity            FROM raw.orders            ORDER BY ingested_at, order_id        """
        df = client.query_df(sql)
    else:        sql = """            SELECT                order_id,                customer_id,                created_at,                ingested_at,                price,                quantity            FROM raw.orders            WHERE ingested_at > {last_ingested_at:DateTime}            ORDER BY ingested_at, order_id        """
        df = client.query_df(            sql,            parameters={"last_ingested_at": last_ingested_at},        )
    if df.empty:        print("No new orders")        return
    df["created_at"] = pd.to_datetime(        df["created_at"],        utc=True,    )
    df["ingested_at"] = pd.to_datetime(        df["ingested_at"],        utc=True,    )
    df["total"] = (        df["price"] * df["quantity"]    )
    columns = [        "order_id",        "customer_id",        "created_at",        "ingested_at",        "total",    ]
    client.insert_df(        "staging.orders",        column_names=columns,        df=df[columns],    )
    print(        f"Processed {len(df)} new rows"    )

The learning sequence is:

1. Connect to ClickHouse.
2. Find the current processing boundary.
3. Build a full-load or incremental query.
4. Read the batch into a DataFrame.
5. Finish successfully when the batch is empty.
6. Convert and transform the selected columns.
7. Select the output schema explicitly.
8. Write the result to Staging.
9. Finish successfully so the next run can use the new boundary.

The function is intentionally compact. A production version should also define data quality checks, rerun behavior, credentials, and the exact watermark guarantees of the source.

Common mistakes#

Reading the entire source on every run#

Full loads are easy to start with, but repeated full scans waste work as the source grows. Move to a watermark or fixed-window strategy when the source and correctness requirements allow it.

Treating an empty batch as an error#

No new rows can be a normal scheduled result. Handle df.empty explicitly and finish the run successfully.

Using inconsistent boundaries#

Mixing <=, >=, and > without a defined convention creates duplicates or gaps. Use half-open windows and document which side of the watermark is included.

Advancing state before a successful write#

If the state moves first and the destination write fails, data may be skipped forever. The processed position must follow successful processing.

Ignoring late arrivals#

A business timestamp may not describe arrival order. A strict watermark can miss records that arrive later with an older timestamp.

Ignoring records that share a timestamp#

Timestamp-only state may be ambiguous when several records have the same value. Use a unique ordering field or a composite watermark when required.

Assuming incremental loading prevents duplicates#

Incremental selection reduces the read range; it does not make destination inserts idempotent. Retries still need a duplicate-prevention strategy.

Retrying non-idempotent inserts#

A failed or partially completed insert may be repeated. Design the target write for safe reruns before enabling automatic retries.

Mixing SQL, Pandas, and orchestration into one unreadable function#

Keep source selection, transformation logic, and scheduling responsibilities distinct enough to test and reason about them separately.

Hardcoding credentials#

Connection settings belong in environment or application configuration, not in committed source code.

Interpolating untrusted SQL values#

String formatting can create unsafe or invalid SQL. Use the driver's parameter binding mechanism for dynamic values.

Adding lookback without understanding it#

A lookback can recover late data, but it also re-reads rows and requires deduplication. Add it because the source behavior requires it, not as a generic substitute for clear state management.

Quick reference#

A batch is a bounded unit of work. Incremental processing makes the boundary explicit so the next run can process only what appeared after the previous successful position.

See also#

Airflow   ↓ every scheduled intervalprocess_new_orders()   ↓new batch   ↓ClickHouse