# How to Implement Near Real-Time Data Ingestion from APIs Using DLT with CDC

> Learn to implement near real-time data ingestion from APIs using DLT with CDC. Capture API changes in minutes with `rest_api_source`, `incremental` modes, and `continuous` refresh for exact data.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: how-to-guide
- Published: 2026-05-30

---

**Use DLT's `rest_api_source` with a `cursor_path` and `incremental="append"` or `"merge"` mode, then set `refresh_mode="continuous"` or `schedule="5m"` in your pipeline to capture API changes within minutes while maintaining exactly-once semantics.**

Delta Live Tables (DLT) provides a declarative framework for building continuous data pipelines on Databricks and other platforms. In the DataTalksClub/data-engineering-zoomcamp repository, you'll find production-ready implementations demonstrating how to combine the `dlt.sources.rest_api` module with Change Data Capture (CDC) patterns to achieve near real-time data ingestion from external REST APIs.

## Understanding the CDC Architecture

A robust CDC implementation for API ingestion requires four coordinated components. First, the **REST API source** handles pagination, authentication, and request filtering. Second, the **cursor mechanism** tracks a monotonic column—such as `updated_at` or an incrementing ID—to identify new or changed records. Third, **incremental table definitions** tell DLT to persist the highest cursor value between runs. Finally, **continuous or frequent scheduling** ensures data freshness while avoiding full reloads.

This architecture eliminates duplicate processing by tracking state in the pipeline. When the pipeline restarts, DLT automatically filters the API request to fetch only rows with cursor values greater than the previous watermark, enabling near real-time latency with minimal overhead.

## Configuring the REST API Source

The foundation of CDC ingestion is the `rest_api_source()` configuration, which defines how to extract data from the endpoint and identify changes. In [`cohorts/2026/workshops/dlt/open_library_pipeline.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2026/workshops/dlt/open_library_pipeline.py), the `open_library_source()` function demonstrates this pattern by specifying a `cursor_path` that acts as the CDC watermark.

Key configuration parameters include:

- **`primary_key`**: Uniquely identifies each record for deduplication and merge operations.
- **`cursor_path`**: The JSON path to a monotonically increasing field that indicates record modification time.
- **`write_disposition`**: Set to `"append"` to preserve historical data or `"merge"` to apply updates.

```python
import dlt
from dlt.sources.rest_api import rest_api_source

def open_library_source(query: str = "harry potter"):
    """REST-API source with pagination and CDC cursor."""
    return rest_api_source({
        "client": {"base_url": "https://openlibrary.org"},
        "resource_defaults": {
            "primary_key": "key",
            "write_disposition": "append",
        },
        "resources": [{
            "name": "books",
            "endpoint": {
                "path": "search.json",
                "params": {"q": query, "limit": 100},
                "data_selector": "docs",
                "paginator": {
                    "type": "offset",
                    "limit": 100,
                    "offset_param": "offset",
                    "limit_param": "limit",
                    "total_path": "numFound",
                },
                "cursor_path": "edition_key",  # Monotonic cursor for CDC

            },
        }],
    })

```

The `cursor_path` value `edition_key` ensures that each pipeline run only fetches books with edition keys greater than the previous run's maximum, creating an efficient incremental ingestion pattern.

## Implementing Append-Only CDC

For immutable data sources where records only append and never change, use the `@dlt.table` decorator with `incremental="append"`. This mode stores the maximum cursor value in the pipeline state and automatically applies it as a filter on subsequent runs.

```python
@dlt.table(
    name="open_library_books",
    primary_key="key",
    incremental="append",  # CDC: only new rows

)
def open_library_books():
    """Incremental table materializing API data."""
    return open_library_source(query="harry potter")

pipeline = dlt.pipeline(
    pipeline_name="open_library_cdc",
    destination="duckdb",
    dataset_name="open_library_data",
    refresh_mode="continuous",  # Near real-time cadence

    progress="log",
)

pipeline.run(open_library_books)

```

As implemented in [`cohorts/2026/workshops/dlt/open_library_pipeline.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2026/workshops/dlt/open_library_pipeline.py), this pattern guarantees exactly-once semantics—no duplicate rows are inserted even if the pipeline retries due to transient failures.

## Handling Mutating Data with Merge CDC

When API records can be updated after initial ingestion, configure the table with `incremental="merge"`. This instructs DLT to generate a `MERGE` statement that inserts new rows and updates existing ones when the primary key matches but the cursor indicates a newer version.

```python
@dlt.table(
    name="api_events",
    primary_key="event_id",
    incremental="merge",  # CDC with upserts

)
def api_events():
    return rest_api_source({
        "client": {"base_url": "https://example.com/api"},
        "resource_defaults": {"write_disposition": "append"},
        "resources": [{
            "name": "events",
            "endpoint": {
                "path": "v1/events",
                "params": {"limit": 500},
                "data_selector": "data",
                "cursor_path": "updated_at",  # ISO-8601 timestamp cursor

            },
        }],
    })

```

The `cursor_path` (`updated_at`) is persisted in the pipeline state, ensuring that each run only processes events modified since the last successful execution.

## Scheduling for Near Real-Time Latency

To achieve sub-minute or minute-level latency, configure the pipeline execution cadence using either continuous mode or explicit scheduling. The `refresh_mode="continuous"` parameter keeps the pipeline running indefinitely, polling the API as fast as resources allow, while `schedule="5m"` executes discrete runs every five minutes.

```python
pipeline = dlt.pipeline(
    pipeline_name="api_events_cdc",
    destination="bigquery",
    dataset_name="events_raw",
    schedule="5m",  # Run every 5 minutes

    progress="log",
)

pipeline.run(api_events)  # DLT handles the incremental watermark

```

Choose continuous mode for event-driven architectures requiring minimal latency, or scheduled intervals for cost-sensitive batch workloads that tolerate five- to ten-minute delays.

## Managing Authentication and Secrets

Secure API authentication is critical for production CDC pipelines. As demonstrated in [`cohorts/2025/workshops/dynamic_load_dlt.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2025/workshops/dynamic_load_dlt.py), store sensitive credentials in DLT's secret store ([`.dlt/secrets.toml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/.dlt/secrets.toml)) and reference them via `os.getenv()` inside your source configuration. This approach prevents API keys from being exposed in source code while allowing the pipeline to rotate tokens without code changes.

## Monitoring Pipeline State

Track the CDC watermark and pipeline health using the inspection utilities shown in [`cohorts/2026/workshops/dlt/analysis.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2026/workshops/dlt/analysis.py). The pipeline state stores the latest cursor value, which you can inspect programmatically to verify that incremental loading is progressing as expected and to debug stalls in the near real-time ingestion flow.

## Summary

- **Use `cursor_path`** in `rest_api_source()` to identify new or modified records for CDC tracking.
- **Choose `incremental="append"`** for immutable event streams or `"merge"` for updatable records requiring upsert logic.
- **Configure `primary_key`** on tables to enable deduplication and merge operations.
- **Set `refresh_mode="continuous"`** or `schedule="5m"` to maintain near real-time data freshness.
- **Reference** [`cohorts/2026/workshops/dlt/open_library_pipeline.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2026/workshops/dlt/open_library_pipeline.py) for complete working examples of CDC-enabled API ingestion.

## Frequently Asked Questions

### What is the difference between incremental append and merge modes in DLT CDC?

**Append mode** tracks the maximum cursor value and inserts only new records, making it ideal for immutable event logs. **Merge mode** generates a `MERGE` statement that updates existing rows when the primary key matches and the cursor indicates a newer version, which is necessary when API records can be modified after creation.

### How does DLT handle exactly-once semantics for API ingestion?

DLT stores the highest cursor value in the pipeline state after each successful run. If a run fails and retries, the pipeline resumes from the last committed watermark, ensuring that no records are duplicated in the destination table even when API calls are retried due to network timeouts.

### Can I implement CDC if the API does not provide a timestamp or incrementing ID?

No—a monotonic cursor field (timestamp, version number, or incrementing ID) is required for CDC. If the API lacks this, you must implement a full refresh pattern or store the entire dataset locally and compute deltas using hash comparisons, though this approach sacrifices the efficiency benefits of true CDC.

### How do I monitor the CDC watermark to verify incremental loading is working?

Inspect the pipeline state using the methods shown in [`cohorts/2026/workshops/dlt/analysis.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2026/workshops/dlt/analysis.py) to retrieve the current cursor value. Compare this against the source API's maximum value to ensure the pipeline is keeping pace with incoming data and not falling behind in near real-time scenarios.