Real-Time Dashboards with MV & CDC: A Deep Dive

As a senior software engineer, I’ve seen firsthand how quickly business demands evolve. The days of waiting for nightly batch reports are long gone. Today, organizations thrive on instant insights, requiring data to be not just available, but actionable, in near real-time. This pursuit of immediacy is particularly critical for operational dashboards, fraud detection, personalized user experiences, and IoT monitoring.

At Khadervali.com, we believe in empowering developers with the knowledge to build resilient, high-performance systems. That’s why today, we’re diving deep into a powerful architectural pattern for constructing real-time dashboards: combining Materialized Views (MVs) with Change Data Capture (CDC). This synergy allows us to achieve low-latency data processing and serve up-to-the-second analytics without crippling our transactional databases.

This article will serve as your comprehensive guide. We’ll explore the ‘why’ behind real-time, dissect CDC and MVs individually, then demonstrate how their combined power can transform your data pipelines. We’ll walk through a practical implementation using PostgreSQL, Debezium, Kafka, and a custom Python consumer, complete with code examples and architectural insights. By the end, you’ll have a clear understanding and the tools to start building your own real-time analytics solutions.

The Imperative of Real-Time: Why Latency Matters

In today’s hyper-connected world, decisions are made in milliseconds. The value of data diminishes rapidly with time. What was a critical insight an hour ago might be obsolete now. Consider these scenarios:

• E-commerce Analytics: A sudden spike in failed transactions could indicate a payment gateway issue. Real-time monitoring allows immediate intervention, saving revenue and customer satisfaction.
• Fraud Detection: Identifying suspicious activity as it happens is crucial. A delay of even a few minutes can allow fraudsters to complete their schemes.
• IoT Device Monitoring: A malfunctioning sensor in a critical industrial setup needs immediate attention. Real-time alerts prevent costly equipment damage or downtime.
• Personalized User Experiences: Recommending products or content based on a user’s *current* browsing session, not their activity from yesterday, significantly improves engagement.
• Logistics and Supply Chain: Tracking goods movement, predicting delays, and optimizing routes in real-time can lead to massive efficiency gains and cost savings.

These examples underscore a fundamental truth: the closer to real-time your data reflects reality, the more effectively you can react, adapt, and innovate. Traditional batch processing, while still valuable for historical analysis and reporting, simply cannot keep pace with these modern demands.

Traditional Approaches and Their Limitations

Before diving into our chosen solution, let’s briefly examine how real-time challenges were (and sometimes still are) addressed, and why those methods often fall short:

1. Direct Database Queries

The simplest approach for a dashboard is to query the operational database directly. While this provides the freshest data, it’s often unsustainable:

• Performance Impact: Complex analytical queries can put a significant load on your transactional database, slowing down critical application operations.
• Query Latency: Even if the database can handle the load, complex queries take time, meaning the “real-time” aspect is compromised by the query execution duration itself.
• Data Model Mismatch: Operational databases are optimized for transactional integrity (OLTP), not analytical queries (OLAP). Joining many tables or performing aggregations can be highly inefficient.

2. Polling

Dashboards often poll the database at regular intervals (e.g., every 5 seconds) to fetch updated data. This reduces the impact of a single query but introduces other issues:

• Inefficiency: Most polls might return no new data, wasting resources.
• Data Staleness: Data is only as fresh as the polling interval. If updates happen between polls, the dashboard remains outdated.
• Increased Load: Frequent polling, even for small datasets, can add up to a significant load over time, especially with many concurrent dashboard users.

3. Batch Processing and ETL

This is the workhorse of traditional data warehousing. Data is extracted, transformed, and loaded (ETL) into a separate data warehouse or data mart, typically overnight or at fixed intervals.

• High Latency: By design, batch processing introduces significant delays (hours to a day).
• Resource Intensive: Batch jobs can consume substantial resources during their execution window.
• Complex Maintenance: Managing batch dependencies, retries, and failures can be cumbersome.

These limitations highlight the need for an architecture that can deliver fresh data with minimal latency, without overburdening source systems, and in a scalable manner. This is precisely where Change Data Capture and Materialized Views come into play.

Understanding Change Data Capture (CDC)

Change Data Capture (CDC) is a set of software design patterns used to determine and track the data that has changed in a database so that action can be taken using the changed data. Instead of periodically querying the entire database or relying on application-level triggers, CDC focuses on capturing changes at their source, typically by reading the database’s transaction log.

How CDC Works

Most relational databases maintain a transaction log (also known as a write-ahead log, WAL, or redo log). This log records every change made to the database (INSERT, UPDATE, DELETE) before the change is applied to the actual data files. CDC tools leverage this log for a non-intrusive, efficient way to capture changes:

1. Log-Based CDC: This is the most common and robust form. A CDC connector or agent reads the database’s transaction log. Since the log is an immutable, ordered record of all changes, this method provides complete, consistent, and low-latency change events without impacting the performance of the source database.
2. Trigger-Based CDC: Involves creating database triggers on tables to record changes into a separate shadow table. This can introduce overhead to the source database and requires careful management of triggers.
3. Timestamp-Based CDC: Involves adding a `last_updated_at` column to tables and periodically querying for rows where this timestamp is newer than the last extraction. This is simpler but can miss deletes and might not capture intermediate changes effectively.

For real-time dashboards, log-based CDC is the gold standard due to its minimal impact, completeness, and low latency.

Benefits of Log-Based CDC

• Low Latency: Changes are captured almost immediately after they are committed to the database.
• Minimal Impact on Source Database: Reading the transaction log is a passive operation that doesn’t add significant load to the operational database.
• Completeness and Consistency: Captures all changes in the order they occurred, ensuring data integrity.
• Decoupling: Separates the concerns of transactional processing from data ingestion for analytics.
• Scalability: Change events can be streamed to a message queue (like Kafka), allowing downstream consumers to scale independently.

Common CDC Tools and Frameworks

• Debezium: An open-source distributed platform for CDC. It provides a set of Kafka Connect connectors that monitor various databases (PostgreSQL, MySQL, SQL Server, MongoDB, Oracle) and stream row-level changes into Kafka topics.
• Fivetran/Airbyte: Managed ETL/CDC services that abstract away much of the complexity.
• Native Database Features: Some databases offer their own CDC mechanisms (e.g., SQL Server Change Data Capture, Oracle GoldenGate).

For our practical example, we’ll focus on Debezium as it’s a powerful, open-source, and widely adopted solution that integrates seamlessly with Kafka.

Understanding Materialized Views (MVs)

A Materialized View is a database object that contains the results of a query. Unlike a regular view, which is just a stored query that gets executed every time it’s accessed, a Materialized View stores the actual data on disk. This pre-computation makes queries against the MV significantly faster, especially for complex aggregations or joins.

How Materialized Views Work

When you create an MV, the database executes the underlying query and stores the result. Subsequent queries against the MV read from this pre-computed result set. The key challenge with MVs is keeping them fresh.

Refreshing Materialized Views

Since MVs store static data, they can become stale as the underlying base tables change. They need to be refreshed to reflect the latest data:

• Full Refresh: The entire query defining the MV is re-executed, and the MV’s data is completely rebuilt. This is simple but can be resource-intensive and time-consuming for large MVs, making it unsuitable for low-latency scenarios.
• Incremental Refresh: Only the changes (inserts, updates, deletes) that occurred in the base tables since the last refresh are applied to the MV. This is much more efficient for frequently changing data but requires more complex logic to track and apply these changes. Many traditional databases have limitations on incremental refresh capabilities, often only supporting specific scenarios or requiring enterprise editions.

Benefits of Materialized Views

• Performance: Dramatically speeds up complex, analytical queries by pre-calculating results.
• Reduced Load on Source System: Once computed, queries against the MV don’t impact the base tables.
• Simplified Queries: Users and applications can query the MV directly, avoiding complex joins or aggregations.
• Network Efficiency: When MVs are replicated or stored closer to consumers, they reduce network traffic to the primary database.

Limitations and Challenges

• Data Staleness: Unless constantly refreshed, MVs can show outdated data.
• Storage Cost: MVs consume disk space, potentially duplicating data.
• Refresh Overhead: The process of refreshing an MV, especially a full refresh, can be resource-intensive and block queries during its execution.
• Complexity for Incremental Updates: Manually managing incremental updates can be complex, requiring careful tracking of changes and transactional consistency.

For our real-time dashboard goal, we need an MV solution that supports *incremental* updates with very low latency. While some databases offer native incremental MVs, we’ll demonstrate a pattern where we manage the “materialized view” ourselves (as a regular table) and update it using CDC events. This gives us finer control and flexibility.

The Synergy: CDC + Materialized Views for Real-Time Dashboards

This is where the magic happens. By combining CDC with Materialized Views, we create a powerful pipeline that continuously updates our aggregated data structures with minimal latency and without hammering our source transactional databases.

The Architectural Blueprint (Diagram in Words)

Imagine a flow of data, originating from your core transactional database, flowing through a series of specialized components, and finally landing in a highly optimized structure ready for dashboard consumption. Here’s how it looks:

+-----------------+       +----------------+       +-------------------+       +-----------------+       +-----------------+
| Operational DB  | ----> | CDC Connector  | ----> | Message Queue     | ----> | Stream Processor| ----> | Materialized    |
| (e.g., PostgreSQL)|       | (e.g., Debezium)|       | (e.g., Kafka)     |       | (e.g., Python   |       | View Store      |
| Source of truth |       | Captures Changes|       | Event Hub         |       | Consumer/Flink) |       | (e.g., Analytics |
|                 |       | (INSERT/UPDATE/ |       |                   |       | Transforms &    |       | DB / PostgreSQL |
|                 |       | DELETE)         |       |                   |       | Applies Updates |       | Table)          |
+-----------------+       +----------------+       +-------------------+       +-----------------+       +-----------------+
                                                                                                                    |
                                                                                                                    |
                                                                                                                    V
                                                                                                            +-----------------+
                                                                                                            | Dashboard/BI    |
                                                                                                            | Tool (Grafana,  |
                                                                                                            | Tableau, etc.)  |
                                                                                                            +-----------------+

Let’s break down each component in this pipeline:

1. Operational Database (Source of Truth): Your primary database (PostgreSQL in our example) where transactional data is generated and updated. This is optimized for OLTP workloads.
2. CDC Connector (Debezium): A lightweight agent that monitors the transaction log of the Operational DB. It captures every change (INSERT, UPDATE, DELETE) at the row level.
3. Message Queue (Kafka): Debezium streams these raw change events into Kafka topics. Kafka acts as a durable, fault-tolerant, and scalable message bus, decoupling the CDC process from downstream consumers. It allows multiple consumers to process the same data stream independently.
4. Stream Processor (Custom Python Consumer): This is the brain of our real-time pipeline. It consumes messages from Kafka, parses the CDC events, and applies the necessary logic to update our Materialized View. This could be a simple Python script, a more robust Kafka Streams application, or even a dedicated stream processing engine like Apache Flink or Spark Streaming for complex aggregations.
5. Materialized View Store (Analytics DB / PostgreSQL Table): A separate database or schema, potentially even the same PostgreSQL instance but a different database, specifically designed to hold the pre-aggregated, denormalized, and optimized Materialized Views. This database is optimized for read-heavy analytical queries (OLAP). In our example, this will be a regular PostgreSQL table that we manage as our MV.
6. Dashboard/BI Tool: Finally, your dashboarding solution (Grafana, Metabase, Tableau, Power BI) connects directly to the Materialized View Store, querying the pre-computed results with blazing speed.

This architecture provides a robust, scalable, and low-latency solution:

• Real-time Data: Changes propagate from the source to the dashboard in seconds or sub-seconds.
• Decoupling: Each component can scale and fail independently. The source database remains untouched by analytical queries.
• Flexibility: The message queue allows for multiple downstream consumers, enabling various analytical applications to leverage the same change stream.
• Performance: Dashboards query highly optimized, pre-aggregated data, ensuring fast response times.

Deep Dive into Implementation Details (with Code Examples)

Let’s get practical. We’ll set up a simple scenario: monitoring product sales in real-time. We’ll have a transactional products table and an orders table. Our goal is to build a dashboard that shows aggregated product sales data (e.g., total quantity sold, total revenue per product) in near real-time.

We’ll use:

• PostgreSQL: As our source operational database and also our Materialized View store.
• Debezium: To capture changes from PostgreSQL.
• Apache Kafka: To stream change events.
• Python: For our custom stream processor to update the Materialized View.

For this walkthrough, we assume you have Docker and Docker Compose installed, which simplifies the setup of PostgreSQL, Kafka, and Debezium Connect.

Step 1: Setting Up PostgreSQL for CDC

First, we need to configure our PostgreSQL database to enable logical decoding, which Debezium uses to read the transaction log. We’ll use Docker Compose for this.

Create a docker-compose.yml file:

# docker-compose.yml
version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - zookeeper

  postgres:
    image: debezium/postgres:15
    hostname: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: sales_db
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: password
      # Enable logical decoding for Debezium
      POSTGRES_INITDB_ARGS: "--encoding=UTF-8 --lc-collate=C --lc-ctype=C"
      PGDATA: /var/lib/postgresql/data/pgdata
    volumes:
      - ./pgdata:/var/lib/postgresql/data/pgdata
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U admin -d sales_db"]
      interval: 5s
      timeout: 5s
      retries: 5

  connect:
    image: debezium/connect:2.4
    hostname: connect
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-status
      # Set replication factor for Kafka Connect internal topics
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
    depends_on:
      - kafka
      - postgres

Save this file and run docker-compose up -d. This will start Zookeeper, Kafka, PostgreSQL, and the Kafka Connect container (where Debezium will run).

Once PostgreSQL is running, connect to it and create your source tables and the “materialized view” table:

-- Connect to sales_db using psql: psql -h localhost -p 5432 -U admin -d sales_db
-- Password: password

-- Enable logical decoding on the database (if not already done via docker-compose)
-- This is usually configured in postgresql.conf, but Debezium's image handles it.
-- For manual setup, you'd add:
-- wal_level = logical
-- max_wal_senders = 10
-- max_replication_slots = 10
-- to postgresql.conf and restart.

-- Create source tables
CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    price DECIMAL(10, 2) NOT NULL
);

CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,
    product_id INT NOT NULL,
    quantity INT NOT NULL,
    order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(50) DEFAULT 'completed',
    FOREIGN KEY (product_id) REFERENCES products(product_id)
);

-- Insert some initial data
INSERT INTO products (name, price) VALUES
('Laptop', 1200.00),
('Mouse', 25.00),
('Keyboard', 75.00);

INSERT INTO orders (product_id, quantity) VALUES
(1, 1),
(2, 2),
(3, 1),
(1, 2);

-- Our "Materialized View" table for real-time sales aggregates
-- This table will be incrementally updated by our Python consumer
CREATE TABLE product_sales_mv (
    product_id INT PRIMARY KEY,
    product_name VARCHAR(255) NOT NULL,
    total_quantity_sold INT DEFAULT 0,
    total_revenue DECIMAL(10, 2) DEFAULT 0.00,
    last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Initialize the MV with current data
INSERT INTO product_sales_mv (product_id, product_name, total_quantity_sold, total_revenue)
SELECT
    p.product_id,
    p.name,
    COALESCE(SUM(o.quantity), 0) AS total_quantity_sold,
    COALESCE(SUM(o.quantity * p.price), 0.00) AS total_revenue
FROM products p
LEFT JOIN orders o ON p.product_id = o.product_id
GROUP BY p.product_id, p.name;

Step 2: Configuring Debezium PostgreSQL Connector

Now, we’ll configure Debezium to monitor our sales_db PostgreSQL database and stream changes from the orders table to Kafka. You interact with Kafka Connect (which Debezium uses) via its REST API.

Execute this command to register the Debezium connector. Make sure the Kafka Connect container (`connect`) is fully up and running before executing this.

curl -X POST -H "Content-Type: application/json" --data '{
    "name": "sales-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "database.hostname": "postgres",
        "database.port": "5432",
        "database.user": "admin",
        "database.password": "password",
        "database.dbname": "sales_db",
        "database.server.name": "sales_server",
        "table.include.list": "public.orders",
        "plugin.name": "pgoutput",
        "publication.autocreate.mode": "filtered",
        "heartbeat.interval.ms": "5000",
        "slot.name": "debezium_slot",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter.schemas.enable": "false"
    }
}' http://localhost:8083/connectors

This command registers a connector named sales-connector. It’s configured to monitor the public.orders table in sales_db. Debezium will create a Kafka topic named sales_server.public.orders (based on database.server.name and table name) and publish change events there.

Step 3: Building the Kafka Consumer & Materialized View Updater (Python)

This is the core logic. Our Python script will connect to Kafka, consume messages from the sales_server.public.orders topic, parse the Debezium event payload, and then apply the corresponding incremental update to our product_sales_mv table.

First, install the necessary Python libraries:

pip install confluent-kafka psycopg2-binary

Now, create a Python file named mv_updater.py:

# mv_updater.py

from confluent_kafka import Consumer, KafkaException

import json

import psycopg2

import sys

import time

# Kafka Consumer configuration

conf = {

    'bootstrap.servers': 'localhost:9093', # Kafka broker exposed on host

    'group.id': 'mv-updater-group',

    'auto.offset.reset': 'earliest',

    'enable.auto.commit': False # We'll commit manually

}
# PostgreSQL connection details for the MV store

DB_HOST = 'localhost' # Connect directly to the host's exposed port

DB_PORT = 5432

DB_NAME = 'sales_db'

DB_USER = 'admin'

DB_PASSWORD = 'password'
def get_db_connection():

    """Establishes and returns a PostgreSQL database connection."""

    conn = psycopg2.connect(

        host=DB_HOST,

        port=DB_PORT,

        database=DB_NAME,

        user=DB_USER,

        password=DB_PASSWORD

    )

    return conn
def update_materialized_view(operation, payload, cursor):

    """

    Applies incremental updates to the product_sales_mv based on CDC events.

    Handles INSERT, UPDATE, and DELETE operations for the 'orders' table.

    """

    try:

        if not payload:

            print("Payload is empty, skipping update.")

            return
        # Extract relevant data based on operation type

        product_id = None

        quantity = None

        old_quantity = None
        if operation == 'c' or operation == 'r': # Create (insert) or Read (snapshot)

            product_id = payload['product_id']

            quantity = payload['quantity']

            print(f"Processing INSERT/SNAPSHOT for order {payload['order_id']}: Product ID {product_id}, Quantity {quantity}")

        elif operation == 'u': # Update

            # 'before' contains the old values, 'after' contains the new values

            old_product_id = payload['before']['product_id']

            old_quantity = payload['before']['quantity']

            product_id = payload['after']['product_id']

            quantity = payload['after']['quantity']

            print(f"Processing UPDATE for order {payload['before']['order_id']}: Old Product ID {old_product_id}, Old Qty {old_quantity} -> New Product ID {product_id}, New Qty {quantity}")
            # Handle cases where product_id changes in an update (rare but possible)

            if