Building Real-Time Data Pipelines with SQLite

Change Data Capture and Stream Processing

SQLite is embedded, lightweight, and file-based. It is not a streaming platform. It does not include built-in replication logs or event streams.

However, SQLite can serve as a reliable source for real-time data pipelines when you implement Change Data Capture (CDC) correctly.

This article explains:

• Trigger-based CDC patterns in SQLite

• Log table streaming architecture

• Integrating SQLite with Kafka or Pulsar

• How to avoid common streaming mistakes

The goal is practical, production-safe design.

What Change Data Capture Means in SQLite

Change Data Capture is the process of detecting and recording data modifications so they can be consumed elsewhere.

In enterprise systems, CDC often reads database transaction logs. SQLite does not expose a logical replication stream.

In SQLite, CDC is typically implemented using:

1. Triggers

2. WAL tailing tools

3. Application-level event publishing

The most predictable method is trigger-based CDC with a log table.

Core Architecture

The reliable SQLite CDC pattern looks like this:

1. Application writes to primary tables

2. SQLite triggers write change events into a cdc_log table

3. A streaming worker reads new entries

4. The worker publishes events to Kafka or Pulsar

5. Processed events are marked or checkpointed

This keeps SQLite simple and streaming logic external.

Step 1: Create a CDC Log Table

CREATE TABLE IF NOT EXISTS cdc_log (
  id            INTEGER PRIMARY KEY AUTOINCREMENT,
  table_name    TEXT NOT NULL,
  operation     TEXT NOT NULL,
  row_id        INTEGER NOT NULL,
  payload       TEXT NOT NULL,
  created_at    TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  processed     INTEGER NOT NULL DEFAULT 0
);

CREATE INDEX IF NOT EXISTS idx_cdc_processed ON cdc_log(processed);

This table stores change events in JSON format.

Step 2: Add Triggers for Change Capture

Example primary table:

CREATE TABLE IF NOT EXISTS users (
  user_id    INTEGER PRIMARY KEY,
  email      TEXT NOT NULL,
  plan       TEXT NOT NULL,
  updated_at TEXT NOT NULL
);

Insert trigger:

CREATE TRIGGER users_after_insert
AFTER INSERT ON users
BEGIN
  INSERT INTO cdc_log (table_name, operation, row_id, payload)
  VALUES (
    ‘users’,
    ‘INSERT’,
    NEW.user_id,
    json_object(
      ‘user_id’, NEW.user_id,
      ‘email’, NEW.email,
      ‘plan’, NEW.plan,
      ‘updated_at’, NEW.updated_at
    )
  );
END;

Update trigger:

CREATE TRIGGER users_after_update
AFTER UPDATE ON users
BEGIN
  INSERT INTO cdc_log (table_name, operation, row_id, payload)
  VALUES (
    ‘users’,
    ‘UPDATE’,
    NEW.user_id,
    json_object(
      ‘user_id’, NEW.user_id,
      ‘email’, NEW.email,
      ‘plan’, NEW.plan,
      ‘updated_at’, NEW.updated_at
    )
  );
END;

Delete trigger:

CREATE TRIGGER users_after_delete
AFTER DELETE ON users
BEGIN
  INSERT INTO cdc_log (table_name, operation, row_id, payload)
  VALUES (
    ‘users’,
    ‘DELETE’,
    OLD.user_id,
    json_object(
      ‘user_id’, OLD.user_id
    )
  );
END;

Now every change produces a structured event.

Step 3: Streaming Worker Design

The worker continuously polls unprocessed rows.

Example Python worker:

import sqlite3
import json
from kafka import KafkaProducer

DB_PATH = “app.db”

producer = KafkaProducer(
    bootstrap_servers=”localhost:9092”,
    value_serializer=lambda v: json.dumps(v).encode(”utf-8”)
)

def main():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row

    while True:
        rows = conn.execute(”“”
            SELECT id, table_name, operation, payload
            FROM cdc_log
            WHERE processed = 0
            ORDER BY id
            LIMIT 100
        “”“).fetchall()

        if not rows:
            continue

        for row in rows:
            event = {
                “table”: row[”table_name”],
                “operation”: row[”operation”],
                “payload”: json.loads(row[”payload”])
            }

            producer.send(”sqlite-events”, event)

            conn.execute(
                “UPDATE cdc_log SET processed = 1 WHERE id = ?”,
                (row[”id”],)
            )

        conn.commit()

if __name__ == “__main__”:
    main()

Install dependency:

pip install kafka-python

This is a simple polling-based stream publisher.

Kafka vs Pulsar Integration

The logic is identical. Only the producer changes.

For Pulsar:

pip install pulsar-client

Producer example:

import pulsar

client = pulsar.Client(”pulsar://localhost:6650”)
producer = client.create_producer(”sqlite-events”)

producer.send(json.dumps(event).encode(”utf-8”))

The CDC design remains the same.

How to Make This Production-Grade

1. Use WAL Mode

PRAGMA journal_mode = WAL;

WAL improves concurrency between writes and streaming reads.

For details on WAL concurrency behavior, see https://www.sqliteforum.com/p/mastering-transactions-and-concurrency

2. Use Batch Processing

Process events in batches, not one at a time. Reduce commit frequency.

3. Use Checkpointing Instead of Boolean Flags

Instead of processed = 0, track last processed ID:

SELECT * FROM cdc_log WHERE id > ? ORDER BY id LIMIT 100;

Store last processed ID externally.

This reduces write amplification.

4. Avoid Large JSON Payloads

Keep payload minimal. Downstream systems can enrich.

5. Protect Against Infinite Growth

Implement log cleanup:

DELETE FROM cdc_log
WHERE processed = 1
AND id < ?;

Or archive older rows.

Alternative CDC Pattern: WAL Tailing

Some tools tail SQLite WAL files directly.

This approach:

• Avoids triggers

• Reads low-level WAL changes

• Requires deeper SQLite internals knowledge

It is more complex and harder to reason about. Trigger-based CDC is easier to control.

What This Pattern Is Not

This does not make SQLite:

• A distributed streaming database

• A message broker

• A replacement for Kafka

SQLite is the event source. Kafka or Pulsar handle distribution and scaling.

When This Architecture Works

Use SQLite CDC when:

• SQLite is embedded in edge devices

• You need to stream changes to a central system

• You want local durability with asynchronous export

• Event throughput is moderate

Do not use this pattern for ultra-high throughput write-heavy systems.

Closing Notes

SQLite can serve as a reliable real-time event source when you:

• Capture changes deterministically

• Stream externally

• Control growth and backpressure

CDC in SQLite is not automatic. It is architectural.

Design it intentionally.

Subscribe Now

If you want more practical, execution-focused SQLite content, subscribe to SQLite Forum.

Upcoming topics include:

• WAL checkpoint tuning for streaming systems

• Exactly-once semantics with SQLite CDC

• Backpressure handling in embedded pipelines

• Designing idempotent consumers

Subscribe to receive new articles directly.