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

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, 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.

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.

@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, 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.

@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.

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, store sensitive credentials in DLT's secret store (.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. 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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}