Real Time Stream Processing

Welcome to the final lesson of our course on optimizing and scaling vector search systems. In previous lessons, we explored dynamic search space reduction to enhance the efficiency of vector search. In this lesson, we will focus on real-time stream processing, a crucial capability for modern applications that require immediate data updates and retrieval.

Real-time stream processing refers to the continuous input, processing, and output of data as it arrives, rather than waiting to process data in large batches. This approach is essential for applications that need to reflect the most current information, such as recommendation engines, live analytics dashboards, and systems that monitor rapidly changing data. In the context of vector search, real-time stream processing ensures that the search index is always up-to-date, allowing users to retrieve the most relevant and recent results.

It’s important to understand how real-time streaming compares to other data ingestion strategies:

• Real-Time Streaming: Data is ingested and indexed as soon as it arrives. This is ideal for use cases where up-to-the-second freshness is critical, such as fraud detection, live personalization, or monitoring systems.
• Batch Updates: Data is collected over a period and then processed in bulk. This approach is often sufficient for applications where slight delays are acceptable, such as nightly analytics, periodic content updates, or offline model training.
• Micro-Batching: A hybrid approach where small batches are processed at short, regular intervals. This can balance the need for freshness with system efficiency.
• Scheduled Syncs: Data is synchronized at fixed intervals (e.g., every hour or day), which is suitable for static or slowly changing datasets.

Choosing the right approach depends on your application’s requirements for data freshness, system complexity, and resource constraints. Real-time streaming is most suitable when immediate data availability is a priority, while batch or scheduled updates may be sufficient for less time-sensitive scenarios.

Before implementing real-time streaming, it is important to set up your environment to support continuous data ingestion and indexing. In a typical vector search system, this involves:

• Loading or preparing the initial set of documents or data points.
• Creating or connecting to a vector index, which will store the vector representations of your data.
• Generating vector embeddings for each document using an embedding function.
• Preparing the data for insertion, including any relevant metadata.

Here’s how you might set up your environment in Python with Qdrant and SentenceTransformers:

import json
from qdrant_client import QdrantClient
from qdrant_client.http import models
from sentence_transformers import SentenceTransformer

# Config
collection_name = "vector-search"
file_path = "./data/corpus.json"

# Load embedding model
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

# Load documents from JSON file
with open(file_path, "r", encoding="utf-8") as f:
    documents = json.load(f)

# Initialize Qdrant client (in-memory for demo)
client = QdrantClient(":memory:")

# Create collection
vector_dim = model.get_sentence_embedding_dimension()
if client.collection_exists(collection_name):
    client.delete_collection(collection_name)

client.create_collection(
    collection_name=collection_name,
    vectors_config=models.VectorParams(size=vector_dim, distance=models.Distance.COSINE)
)

Explanation:

• We import the necessary libraries for working with Qdrant and SentenceTransformers.
• The embedding model is loaded using SentenceTransformer.
• Documents are loaded from a JSON file.
• We initialize a Qdrant client (using in-memory storage for demonstration).
• If a collection with the same name exists, it is deleted to start fresh.
• A new collection is created with the appropriate vector dimension and cosine distance metric.

With the environment ready, let’s implement real-time data streaming. This involves continuously inserting new data into the vector index in small batches, along with metadata such as timestamps.

import time
from datetime import datetime

def stream_data_in_batches(documents, batch_size=10):
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        points = []

        for j, doc in enumerate(batch):
            timestamp = time.time()
            embedding = model.encode(doc["content"]).tolist()

            point = models.PointStruct(
                id=doc.get("id", i + j),
                vector=embedding,
                payload={
                    "id": doc.get("id", i + j),
                    "title": doc.get("title", ""),
                    "category": doc.get("category", "Unknown"),
                    "date": doc.get("date", ""),
                    "tags": ", ".join(doc.get("tags", [])),
                    "source": "streaming",
                    "timestamp": timestamp
                }
            )
            points.append(point)

            formatted = datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
            print(f"Prepared document ID: {doc.get('id', i + j)}, Timestamp: {formatted}")

        client.upsert(collection_name=collection_name, points=points)
        time.sleep(0.1)  # Simulate delay between batches

# Start streaming
stream_data_in_batches(documents)
print("Real-time streaming of new documents completed.")

Explanation:

• The stream_data_in_batches function processes the documents in small batches (default size 10).
• For each document in the batch:
  • The current timestamp is recorded.
  • The document content is encoded into a vector embedding.
  • A PointStruct is created with the embedding and metadata (including timestamp, title, category, etc.).
  • A print statement outputs the document ID and formatted timestamp for monitoring.
• After preparing the batch, the points are upserted (inserted or updated) into the Qdrant collection.
• A short delay (time.sleep(0.1)) simulates real-time data arrival.
• After all batches are processed, a completion message is printed.

Example Output:

Prepared document ID: 1, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 2, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 3, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 4, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 5, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 6, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 7, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 8, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 9, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 10, Timestamp: 2024-06-10 14:23:45
Prepared document ID: 11, Timestamp: 2024-06-10 14:23:46
...
Real-time streaming of new documents completed.

This output shows each document being prepared and streamed with its timestamp, providing real-time feedback during the streaming process.

After implementing real-time streaming, it is important to monitor and optimize your process to ensure efficiency and reliability. Consider the following strategies:

1. Performance Monitoring: Track batch insertion times to detect bottlenecks. You can extend the print statements to log how long each batch takes to process.
2. Error Handling: Add error handling for issues such as network interruptions or malformed data. For example, wrap the upsert call in a try-except block and log any errors.
3. Scalability: Ensure the system can handle higher throughput by adjusting batch sizes, scaling hardware, or using distributed deployments.

Example: Adding Batch Timing and Error Handling

def stream_data_in_batches(documents, batch_size=10):
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        points = []
        batch_start = time.time()

        for j, doc in enumerate(batch):
            timestamp = time.time()
            embedding = model.encode(doc["content"]).tolist()
            point = models.PointStruct(
                id=doc.get("id", i + j),
                vector=embedding,
                payload={
                    "id": doc.get("id", i + j),
                    "title": doc.get("title", ""),
                    "category": doc.get("category", "Unknown"),
                    "date": doc.get("date", ""),
                    "tags": ", ".join(doc.get("tags", [])),
                    "source": "streaming",
                    "timestamp": timestamp
                }
            )
            points.append(point)
            formatted = datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
            print(f"Prepared document ID: {doc.get('id', i + j)}, Timestamp: {formatted}")

        try:
            client.upsert(collection_name=collection_name, points=points)
            batch_end = time.time()
            print(f"Batch {i//batch_size + 1} inserted in {batch_end - batch_start:.3f} seconds.")
        except Exception as e:
            print(f"Error inserting batch {i//batch_size + 1}: {e}")

        time.sleep(0.1)

Example Output:

Batch 1 inserted in 0.215 seconds.
Batch 2 inserted in 0.198 seconds.
...

In this lesson, we explored the concept of real-time stream processing and its application in vector search systems. By continuously inserting new data into the index, you can ensure that your search system remains up-to-date and responsive to the latest information. We reviewed how to set up your environment, prepare your data, and implement real-time streaming using Qdrant, with explanations and example outputs to clarify each step. As you move on to practice exercises, experiment with different data sets and streaming scenarios to observe the impact on performance. We also discussed when real-time streaming is most appropriate compared to batch or hybrid approaches, helping you choose the best strategy for your use case.