Skip to content

readme.md

Latest commit

 

History

History
439 lines (280 loc) · 8.86 KB

readme.md

File metadata and controls

439 lines (280 loc) · 8.86 KB
Raw

⚡ Real-Time Event-Driven RAG Pipeline

Production-inspired event-driven vector indexing using PostgreSQL CDC, Debezium, Apache Kafka, PySpark Structured Streaming, Ollama, and Qdrant.

Python Spark Kafka Postgres Debezium Qdrant Ollama

📖 Overview

Traditional RAG systems typically rely on scheduled ETL jobs to regenerate embeddings, introducing synchronization delays and unnecessary recomputation. This project demonstrates an event-driven alternative where every database mutation is captured directly from PostgreSQL's Write-Ahead Log (WAL) and propagated through a streaming pipeline, enabling near real-time semantic indexing.

🚀 Highlights

  • Real-time Change Data Capture (CDC) using PostgreSQL WAL
  • Event-driven architecture with Debezium and Apache Kafka
  • Continuous processing using PySpark Structured Streaming
  • Local embedding generation with Ollama (nomic-embed-text)
  • Automatic synchronization with Qdrant Vector Database
  • Fully local deployment optimized for Apple Silicon
  • No external LLM or embedding APIs required

🏗️ Architecture

Note: Architecture diagram can be added here.

             PostgreSQL
                  │
      Logical Replication (WAL)
                  │
                  ▼
             Debezium CDC
                  │
                  ▼
            Apache Kafka
                  │
                  ▼
     PySpark Structured Streaming
                  │
      Parse → Embed → foreachBatch
                  │
                  ▼
       Ollama (nomic-embed-text)
                  │
                  ▼
              Qdrant
                  │
                  ▼
          Semantic Search

⚙️ Pipeline Overview

Source Database (PostgreSQL)

Configured with:

wal_level = logical

Acts as the transactional source of truth.

CDC Engine (Debezium)

Continuously monitors PostgreSQL's Write-Ahead Log (WAL) and converts row-level database mutations into Kafka events without polling the database.

Event Broker (Apache Kafka)

Acts as the immutable messaging layer that decouples producers from downstream consumers.

Stream Processing (PySpark Structured Streaming)

Consumes CDC events from Kafka and performs:

  • JSON parsing
  • Metadata removal
  • Payload extraction
  • Micro-batch processing
  • Embedding generation
  • Vector upsert into Qdrant

Local AI (Ollama)

Runs the lightweight nomic-embed-text embedding model locally to generate 768-dimensional embeddings without relying on external APIs.

Vector Database (Qdrant)

Stores vector embeddings and enables low-latency semantic retrieval.

📁 Repository Structure

.
├── main/
│   └── spark_rag_consumer.py     # Spark streaming job and Qdrant sink
│
├── search/
│   └── search.py                 # Semantic search CLI
│
├── docker-compose.yml            # PostgreSQL, Kafka, Debezium, Qdrant
├── README.md
└── .gitignore

🔄 Data Flow

INSERT INTO PostgreSQL

        │

        ▼

 Write-Ahead Log (WAL)

        │

        ▼

    Debezium CDC

        │

        ▼

     Kafka Topic

        │

        ▼

Spark Structured Streaming

        │

        ▼

Generate Embeddings

        │

        ▼

Qdrant

        │

        ▼

Semantic Search

💡 Engineering Decisions

Why Change Data Capture?

Instead of periodically polling PostgreSQL using queries like:

SELECT *
FROM articles
WHERE updated_at > ...

Debezium reads directly from PostgreSQL's Write-Ahead Log (WAL), enabling near real-time synchronization.

Benefits

  • Lower database load
  • Low-latency event propagation
  • Event-driven synchronization
  • Automatic capture of every database mutation

Why Apache Kafka?

Kafka serves as the event backbone of the architecture, decoupling producers and consumers while allowing additional downstream systems to subscribe without impacting the source database.

Why PySpark Structured Streaming?

Although this workload could be processed by a lightweight Kafka consumer, Spark Structured Streaming was intentionally chosen to demonstrate production-grade streaming patterns such as:

  • Fault tolerance
  • Checkpointing
  • Micro-batch execution
  • Horizontal scalability

The architecture can scale to higher event throughput with minimal code changes.

Why Ollama?

Embedding generation runs entirely on the local machine.

Benefits

  • No API cost
  • Low latency
  • Offline execution
  • Improved data privacy

Why foreachBatch Instead of a Spark UDF?

Embedding generation is executed inside the foreachBatch sink instead of a Spark UDF. This avoids Python serialization (cloudpickle) issues commonly encountered when network-heavy dependencies such as requests are executed inside distributed Spark workers.

Memory Optimization

The Spark application is intentionally configured with:

Driver Memory     : 1 GB
Executor Memory   : 1 GB

allowing the complete infrastructure to run comfortably on an Apple Silicon laptop with 8 GB unified memory.

⚙️ Prerequisites

  • macOS (Apple Silicon recommended)
  • Docker Desktop
  • Python 3.10+
  • OpenJDK 17
  • Ollama

🚀 Setup

1. Start Infrastructure

docker compose up -d

2. Pull the Embedding Model

ollama pull nomic-embed-text

3. Create the Python Environment

python3 -m venv venv

source venv/bin/activate

pip install pyspark==3.5.1 requests qdrant-client

4. macOS Networking Fix

export _JAVA_OPTIONS="-Djava.net.preferIPv4Stack=true"

export SPARK_LOCAL_IP=127.0.0.1

5. Start the Streaming Consumer

python main/spark_rag_consumer.py

🧪 Verify the Pipeline

Insert a record into PostgreSQL.

INSERT INTO articles (title, content)
VALUES (
    'Event Streaming',
    'Kafka processes data streams in real-time with low latency.'
);

The record flows through:

PostgreSQL

↓

Debezium

↓

Kafka

↓

Spark Structured Streaming

↓

Ollama

↓

Qdrant

↓

Semantic Search

Run the semantic search client.

python search/search.py

📊 Sample Streaming Logs

The streaming consumer initializes the Spark session, consumes CDC events from Kafka, generates embeddings using Ollama, and continuously pushes vectors into Qdrant through Spark micro-batches.

26/07/19 14:19:19 WARN NativeCodeLoader:
Unable to load native-hadoop library for your platform...
using builtin-java classes where applicable.

Setting default log level to "WARN".

26/07/19 14:19:21 WARN ResolveWriteToStream:
Temporary checkpoint location created.

26/07/19 14:19:21 WARN ResolveWriteToStream:
spark.sql.adaptive.enabled is not supported in
streaming DataFrames/Datasets and will be disabled.

🚀 Batch 1: Embedded & pushed 1 vectors to Qdrant!

🚀 Batch 2: Embedded & pushed 4 vectors to Qdrant!

The logs confirm that Spark Structured Streaming successfully consumes Kafka events, generates embeddings locally, and incrementally synchronizes vectors into Qdrant.

🔍 Sample Search Output

Run the following command after inserting records into PostgreSQL:

python search/search.py

Search Output

🔍 Searching for: 'What are the benefits of real-time data streaming over batch?'

🎯 Match Score: 0.6827
📌 Title: AI in Data Engineering
📄 Content: Streaming CDC architectures reduce latency compared to batch ETL pipelines.
--------------------------------------------------
🎯 Match Score: 0.6200
📌 Title: The Evolution of CDC
📄 Content: Change Data Capture ensures that downstream systems stay in perfect sync with the primary database by streaming transaction logs.
--------------------------------------------------

This verifies that newly inserted PostgreSQL records become searchable shortly after flowing through the CDC → Kafka → Spark → Ollama → Qdrant pipeline.

📈 Future Improvements

  • Support UPDATE and DELETE events
  • Batch embedding generation
  • Metadata filtering
  • Hybrid keyword + vector search
  • FastAPI search service
  • Kubernetes deployment
  • Monitoring with Prometheus and Grafana

👨‍💻 Author

Hritik Maheshwari

Data Engineer | Streaming Systems | Distributed Data Platforms | AI Infrastructure