DEV Community

Phil IPS
Phil IPS

Posted on

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

#qdrant #pinecone #weaviate #tutorial

Vector Databases Deep Dive: Qdrant, Pinecone, Weaviate & Milvus Compared

Everything you need to choose, deploy, and optimize a vector database for production RAG and AI applications.

Introduction

Vector databases are the backbone of modern AI applications — RAG, semantic search, recommendations, anomaly detection. But they're not all the same.

Why Vector Databases?

Traditional DB Vector DB
Exact match (B-tree) Approximate Nearest Neighbor (ANN)
Scalar values High-dimensional vectors (768-4096 dims)
SQL queries Distance metrics (cosine, dot, L2)
Millions of rows Billions of vectors
ACID transactions Eventual consistency (usually)

Core Concepts

Vector Similarity Search 

# similarity_basics.py
import numpy as np

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def dot_product(a: np.ndarray, b: np.ndarray) -> float:
    return np.dot(a, b)

def l2_distance(a: np.ndarray, b: np.ndarray) -> float:
    return np.linalg.norm(a - b)

# Normalized vectors: cosine == dot product
# For cosine, always normalize embeddings first!
Enter fullscreen mode Exit fullscreen mode

ANN Algorithms

Algorithm Principle Build Time Search Speed Accuracy Memory
HNSW Hierarchical graph Medium Very Fast High High
IVF Inverted file (clustering) Fast Fast Medium Medium
PQ Product quantization Medium Very Fast Medium Low
DiskANN Disk-based graph Slow Fast High Very Low
Flat Brute force None Slow Perfect High

HNSW is the default for most production use cases.

Database Comparison

Quick Comparison

Feature Qdrant Pinecone Weaviate Milvus Chroma
License Apache 2.0 Proprietary BSD-3 Apache 2.0 Apache 2.0
Self-hosted ✅ ❌ ✅ ✅ ✅
Managed Cloud ✅ ✅ ✅ ✅ (Zilliz) ✅
Filtering Excellent Good Excellent Excellent Good
Hybrid Search ✅ ✅ ✅ ✅ ⚠️
Multi-tenancy ✅ ✅ ✅ ✅ ❌
Replication ✅ Managed ✅ ✅ ❌
GPU Support ✅ ❌ ❌ ✅ ❌
Best For Production, self-host Serverless, ease of use Graph + vector, GraphQL Massive scale, GPU Prototyping, local dev

Detailed: Qdrant (Recommended for Self-Hosted) 

# qdrant_complete.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue, MatchAny,
    SearchParams, HnswConfigDiff, QuantizationConfig,
    ScalarQuantization, ScalarQuantizationConfig,
    OptimizersConfigDiff, CollectionStatus
)
from sentence_transformers import SentenceTransformer
import uuid

class QdrantVectorDB:
    def __init__(self, url: str = "http://localhost:6333", api_key: str = None):
        self.client = QdrantClient(url=url, api_key=api_key)
        self.encoder = SentenceTransformer("BAAI/bge-large-en-v1.5")

    def create_collection(self, name: str, vector_size: int = 1024):
        """Create optimized collection for production."""
        self.client.create_collection(
            collection_name=name,
            vectors_config=VectorParams(
                size=vector_size,
                distance=Distance.COSINE,
                hnsw_config=HnswConfigDiff(
                    m=16,              # Connections per layer (16-64)
                    ef_construct=100   # Build accuracy (100-200)
                ),
                quantization_config=QuantizationConfig(
                    scalar=ScalarQuantization(
                        type=ScalarQuantizationConfig(
                            quantile=0.99,
                            always_ram=True
                        )
                    )
                ),
                on_disk=False  # Keep hot vectors in RAM
            ),
            optimizers_config=OptimizersConfigDiff(
                default_segment_number=2,
                max_segment_size=20000,
                memmap_threshold=20000,
                indexing_threshold=10000,
                flush_interval_sec=5
            )
        )

    def upsert_documents(self, collection: str, documents: list[dict]):
        """Batch upsert with embeddings."""
        texts = [d["text"] for d in documents]
        vectors = self.encoder.encode(texts, show_progress_bar=True)

        points = [
            PointStruct(
                id=str(uuid.uuid4()),
                vector=vec.tolist(),
                payload={
                    "text": doc["text"],
                    "metadata": doc.get("metadata", {}),
                    "timestamp": doc.get("timestamp", "")
                }
            )
            for vec, doc in zip(vectors, documents)
        ]

        self.client.upsert(collection_name=collection, points=points)

    def search(
        self,
        collection: str,
        query: str,
        limit: int = 10,
        filters: dict = None,
        score_threshold: float = 0.7
    ) -> list[dict]:
        query_vector = self.encoder.encode([query])[0]

        query_filter = None
        if filters:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
            query_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=collection,
            query_vector=query_vector.tolist(),
            limit=limit,
            query_filter=query_filter,
            score_threshold=score_threshold,
            search_params=SearchParams(
                hnsw_ef=128,  # Search accuracy (higher = more accurate)
                exact=False
            )
        )

        return [
            {
                "id": r.id,
                "score": r.score,
                "text": r.payload["text"],
                "metadata": r.payload["metadata"]
            }
            for r in results
        ]

    def hybrid_search(
        self,
        collection: str,
        query: str,
        query_vector: list[float] = None,
        limit: int = 10,
        sparse_weight: float = 0.3
    ) -> list[dict]:
        """Hybrid dense + sparse search (requires sparse vectors)."""
        # Requires collection with sparse vectors enabled
        # This is simplified - full hybrid needs sparse vector config
        if query_vector is None:
            query_vector = self.encoder.encode([query])[0].tolist()

        # Use prefetch for hybrid
        from qdrant_client.models import Prefetch, FusionQuery

        results = self.client.query_points(
            collection_name=collection,
            prefetch=[
                Prefetch(
                    query=query_vector,
                    using="dense",
                    limit=limit * 2
                )
            ],
            query=FusionQuery(fusion="rrf"),  # Reciprocal Rank Fusion
            limit=limit
        )

        return [
            {"id": r.id, "score": r.score, "text": r.payload["text"]}
            for r in results.points
        ]

    def delete_by_filter(self, collection: str, filters: dict):
        """Delete points matching filter."""
        conditions = [
            FieldCondition(key=k, match=MatchValue(value=v))
            for k, v in filters.items()
        ]
        self.client.delete(
            collection_name=collection,
            points_selector=Filter(must=conditions)
        )

    def scroll_all(self, collection: str, limit: int = 100):
        """Iterate all points."""
        offset = None
        while True:
            points, offset = self.client.scroll(
                collection_name=collection,
                limit=limit,
                offset=offset,
                with_payload=True,
                with_vectors=False
            )
            for p in points:
                yield p
            if offset is None:
                break
Enter fullscreen mode Exit fullscreen mode

Detailed: Pinecone (Serverless, Managed) 

# pinecone_complete.py
from pinecone import Pinecone, ServerlessSpec
from sentence_transformers import SentenceTransformer
import uuid

class PineconeVectorDB:
    def __init__(self, api_key: str, environment: str = "us-east-1"):
        self.pc = Pinecone(api_key=api_key)
        self.encoder = SentenceTransformer("BAAI/bge-large-en-v1.5")
        self.environment = environment

    def create_index(self, name: str, dimension: int = 1024, metric: str = "cosine"):
        if name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=name,
                dimension=dimension,
                metric=metric,
                spec=ServerlessSpec(
                    cloud="aws",
                    region=self.environment
                )
            )
        self.index = self.pc.Index(name)

    def upsert_documents(self, documents: list[dict], namespace: str = ""):
        texts = [d["text"] for d in documents]
        vectors = self.encoder.encode(texts)

        vectors_to_upsert = [
            {
                "id": str(uuid.uuid4()),
                "values": vec.tolist(),
                "metadata": {
                    "text": doc["text"],
                    **doc.get("metadata", {})
                }
            }
            for vec, doc in zip(vectors, documents)
        ]

        # Batch upsert (100 at a time)
        for i in range(0, len(vectors_to_upsert), 100):
            batch = vectors_to_upsert[i:i+100]
            self.index.upsert(vectors=batch, namespace=namespace)

    def search(
        self,
        query: str,
        limit: int = 10,
        namespace: str = "",
        filters: dict = None
    ) -> list[dict]:
        query_vector = self.encoder.encode([query])[0].tolist()

        filter_dict = None
        if filters:
            filter_dict = {k: {"$eq": v} for k, v in filters.items()}

        results = self.index.query(
            vector=query_vector,
            top_k=limit,
            namespace=namespace,
            filter=filter_dict,
            include_metadata=True
        )

        return [
            {
                "id": r.id,
                "score": r.score,
                "text": r.metadata.pop("text", ""),
                "metadata": r.metadata
            }
            for r in results.matches
        ]
Enter fullscreen mode Exit fullscreen mode

Detailed: Weaviate (GraphQL, Hybrid Search) 

# weaviate_complete.py
import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
from sentence_transformers import SentenceTransformer

class WeaviateVectorDB:
    def __init__(self, url: str, api_key: str = None):
        auth = Auth.api_key(api_key) if api_key else None
        self.client = weaviate.connect_to_wcs(
            cluster_url=url,
            auth_credentials=auth
        )
        self.encoder = SentenceTransformer("BAAI/bge-large-en-v1.5")

    def create_collection(self, name: str):
        self.client.collections.create(
            name=name,
            vectorizer_config=Configure.Vectorizer.none(),  # Bring your own vectors
            properties=[
                Property(name="text", data_type=DataType.TEXT),
                Property(name="source", data_type=DataType.TEXT),
                Property(name="category", data_type=DataType.TEXT),
            ],
            vector_index_config=Configure.VectorIndex.hnsw(
                distance_metric=Configure.VectorIndex.DistanceMetric.COSINE
            )
        )

    def upsert_documents(self, collection: str, documents: list[dict]):
        coll = self.client.collections.get(collection)

        with coll.batch.dynamic() as batch:
            for doc in documents:
                vector = self.encoder.encode(doc["text"])
                batch.add_object(
                    properties={
                        "text": doc["text"],
                        "source": doc.get("metadata", {}).get("source", ""),
                        "category": doc.get("metadata", {}).get("category", "")
                    },
                    vector=vector.tolist()
                )

    def hybrid_search(
        self,
        collection: str,
        query: str,
        limit: int = 10,
        alpha: float = 0.5,  # 0 = keyword, 1 = vector
        filters: dict = None
    ) -> list[dict]:
        coll = self.client.collections.get(collection)
        query_vector = self.encoder.encode(query).tolist()

        where_filter = None
        if filters:
            where_filter = weaviate.classes.query.Filter.by_property("category").equal(
                filters.get("category", "")
            )

        results = coll.query.hybrid(
            query=query,
            vector=query_vector,
            alpha=alpha,
            limit=limit,
            where=where_filter,
            return_metadata=weaviate.classes.query.MetadataQuery(score=True)
        )

        return [
            {
                "id": r.uuid,
                "score": r.metadata.score,
                "text": r.properties["text"],
                "metadata": {k: v for k, v in r.properties.items() if k != "text"}
            }
            for r in results.objects
        ]
Enter fullscreen mode Exit fullscreen mode

Production Deployment

Qdrant on Kubernetes 

# k8s/qdrant.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: qdrant
spec:
  serviceName: qdrant
  replicas: 3
  selector:
    matchLabels:
      app: qdrant
  template:
    metadata:
      labels:
        app: qdrant
    spec:
      containers:
      - name: qdrant
        image: qdrant/qdrant:v1.10.0
        ports:
        - containerPort: 6333
          name: http
        - containerPort: 6334
          name: grpc
        env:
        - name: QDRANT__SERVICE__HTTP_PORT
          value: "6333"
        - name: QDRANT__SERVICE__GRPC_PORT
          value: "6334"
        - name: QDRANT__STORAGE__STORAGE_PATH
          value: "/qdrant/storage"
        volumeMounts:
        - name: storage
          mountPath: /qdrant/storage
        resources:
          requests:
            memory: "4Gi"
            cpu: "2000m"
          limits:
            memory: "8Gi"
            cpu: "4000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 6333
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 6333
          initialDelaySeconds: 10
          periodSeconds: 5
  volumeClaimTemplates:
  - metadata:
      name: storage
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: fast-ssd
      resources:
        requests:
          storage: 100Gi
---
apiVersion: v1
kind: Service
metadata:
  name: qdrant
spec:
  selector:
    app: qdrant
  ports:
  - port: 6333
    targetPort: 6333
    name: http
  - port: 6334
    targetPort: 6334
    name: grpc
Enter fullscreen mode Exit fullscreen mode

Docker Compose (Local Dev) 

# docker-compose.vector.yml
version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:v1.10
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    environment:
      - QDRANT__SERVICE__HTTP_PORT=6333
    healthcheck:
      test: ["CMD", "wget", "-q", "--spider", "http://localhost:6333/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  weaviate:
    image: cr.weaviate.io/semitechnologies/weaviate:1.25
    ports:
      - "8080:8080"
    environment:
      - QUERY_DEFAULTS_LIMIT=25
      - AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED=true
      - PERSISTENCE_DATA_PATH=/var/lib/weaviate
      - DEFAULT_VECTORIZER_MODULE=none
      - ENABLE_MODULES=text2vec-openai
    volumes:
      - weaviate_data:/var/lib/weaviate
    healthcheck:
      test: ["CMD", "wget", "-q", "--spider", "http://localhost:8080/v1/.well-known/ready"]
      interval: 10s
      timeout: 5s
      retries: 5

  milvus:
    image: milvusdb/milvus:v2.4.0
    command: ["milvus", "run", "standalone"]
    ports:
      - "19530:19530"
      - "9091:9091"
    volumes:
      - milvus_data:/var/lib/milvus
    environment:
      - ETCD_ENDPOINTS=etcd:2379
      - MINIO_ADDRESS=minio:9000
    depends_on:
      - etcd
      - minio

  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    command: etcd -advertise-client-urls=http://etcd:2379 -listen-client-urls=http://0.0.0.0:2379
    volumes:
      - etcd_data:/etcd

  minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    command: server /data --console-address ":9001"
    environment:
      - MINIO_ROOT_USER=minioadmin
      - MINIO_ROOT_PASSWORD=minioadmin
    volumes:
      - minio_data:/data
    ports:
      - "9000:9000"
      - "9001:9001"

volumes:
  qdrant_data:
  weaviate_data:
  milvus_data:
  etcd_data:
  minio_data:
Enter fullscreen mode Exit fullscreen mode

Performance Optimization

Index Tuning 

# optimization.py

def optimize_qdrant_for_latency(client, collection: str):
    """Optimize for low latency (real-time search)."""
    client.update_collection(
        collection_name=collection,
        optimizer_config=OptimizersConfigDiff(
            indexing_threshold=0,  # Index immediately
            flush_interval_sec=1,  # Flush frequently
        ),
        hnsw_config=HnswConfigDiff(
            ef_construct=200,  # Higher build accuracy
            m=32               # More connections
        )
    )

def optimize_qdrant_for_throughput(client, collection: str):
    """Optimize for high throughput (batch indexing)."""
    client.update_collection(
        collection_name=collection,
        optimizer_config=OptimizersConfigDiff(
            indexing_threshold=20000,
            flush_interval_sec=30,
            max_segment_size=50000
        ),
        hnsw_config=HnswConfigDiff(
            ef_construct=100,
            m=16
        )
    )

def optimize_for_memory(client, collection: str):
    """Optimize for memory efficiency."""
    client.update_collection(
        collection_name=collection,
        vectors_config=VectorParams(
            size=1024,
            distance=Distance.COSINE,
            quantization_config=QuantizationConfig(
                scalar=ScalarQuantization(
                    type=ScalarQuantizationConfig(
                        quantile=0.99,
                        always_ram=False  # Keep quantized on disk
                    )
                )
            ),
            on_disk=True  # Store original vectors on disk
        )
    )
Enter fullscreen mode Exit fullscreen mode

Batching Strategies 

# batching.py
import asyncio
from typing import AsyncGenerator

async def batched_upsert(
    client,
    collection: str,
    documents: list[dict],
    batch_size: int = 100,
    max_concurrent: int = 4
):
    """Concurrent batched upsert for maximum throughput."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def upsert_batch(batch: list[dict]):
        async with semaphore:
            # Convert to points
            points = [create_point(doc) for doc in batch]
            # Use async client if available
            await client.upsert_async(collection_name=collection, points=points)

    # Create batches
    batches = [
        documents[i:i+batch_size]
        for i in range(0, len(documents), batch_size)
    ]

    # Execute concurrently
    await asyncio.gather(*[upsert_batch(b) for b in batches])

# Optimal batch sizes by DB:
BATCH_SIZES = {
    "qdrant": 100,      # 100-500 points per request
    "pinecone": 100,    # Max 100 per request (serverless)
    "weaviate": 100,    # Batch API
    "milvus": 1000,     # Larger batches OK
    "chroma": 1000,     # Larger batches OK
}
Enter fullscreen mode Exit fullscreen mode

Monitoring & Observability

Key Metrics 

# monitoring.py
from dataclasses import dataclass
from datetime import datetime
import time

@dataclass
class VectorDBMetrics:
    timestamp: str
    operation: str  # upsert, search, delete
    latency_ms: float
    batch_size: int
    vectors_processed: int
    error: str = None

class VectorDBMonitor:
    def __init__(self):
        self.metrics = []

    def record_operation(self, operation: str, batch_size: int, 
                         vectors: int, latency_ms: float, error: str = None):
        self.metrics.append(VectorDBMetrics(
            timestamp=datetime.utcnow().isoformat(),
            operation=operation,
            latency_ms=latency_ms,
            batch_size=batch_size,
            vectors_processed=vectors,
            error=error
        ))

    def get_percentiles(self, operation: str, percentile: float = 0.95) -> float:
        ops = [m.latency_ms for m in self.metrics if m.operation == operation and not m.error]
        if not ops:
            return 0
        ops.sort()
        idx = int(len(ops) * percentile)
        return ops[min(idx, len(ops) - 1)]

    def get_throughput(self, operation: str, window_sec: int = 60) -> float:
        now = time.time()
        cutoff = now - window_sec
        # Assuming metrics have timestamps...
        pass
Enter fullscreen mode Exit fullscreen mode

Health Checks 

# health_checks.py
def check_qdrant_health(client) -> dict:
    """Comprehensive health check."""
    try:
        # Cluster info
        cluster = client.get_cluster_info()

        # Collections status
        collections = client.get_collections()

        # Test search
        start = time.time()
        client.search(
            collection_name="test",
            query_vector=[0.1] * 1024,
            limit=1
        )
        search_latency = (time.time() - start) * 1000

        return {
            "status": "healthy",
            "cluster": cluster,
            "collections": len(collections.collections),
            "search_latency_ms": search_latency,
            "timestamp": datetime.utcnow().isoformat()
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "error": str(e),
            "timestamp": datetime.utcnow().isoformat()
        }
Enter fullscreen mode Exit fullscreen mode

Migration Between Databases 

# migration.py
from qdrant_client import QdrantClient
from pinecone import Pinecone

def migrate_qdrant_to_pinecone(
    qdrant_url: str,
    qdrant_collection: str,
    pinecone_api_key: str,
    pinecone_index: str
):
    qdrant = QdrantClient(url=qdrant_url)
    pinecone = Pinecone(api_key=pinecone_api_key)
    index = pinecone.Index(pinecone_index)

    # Scroll all points from Qdrant
    offset = None
    total = 0

    while True:
        points, offset = qdrant.scroll(
            collection_name=qdrant_collection,
            limit=1000,
            offset=offset,
            with_vectors=True,
            with_payload=True
        )

        if not points:
            break

        # Convert to Pinecone format
        vectors = [
            {
                "id": str(p.id),
                "values": p.vector,
                "metadata": {"text": p.payload.get("text", ""), **p.payload.get("metadata", {})}
            }
            for p in points
        ]

        # Upsert to Pinecone
        index.upsert(vectors=vectors)
        total += len(vectors)
        print(f"Migrated {total} vectors...")

        if offset is None:
            break

    print(f"Migration complete: {total} vectors")
Enter fullscreen mode Exit fullscreen mode

Cost Optimization

Self-Hosted vs Managed

Scale Self-Hosted (Qdrant) Pinecone Serverless Zilliz Cloud (Milvus)
1M vectors $50/mo (1x r6g.xlarge) $70/mo $60/mo
10M vectors $200/mo (3x r6g.xlarge) $700/mo $500/mo
100M vectors $1500/mo (cluster) $7000/mo $4000/mo
1B vectors $10000/mo (cluster) Not recommended $30000/mo

Self-hosted wins at scale. Managed wins for convenience.

Optimization Tips 

# cost_optimization.py

def reduce_vector_dimension(embeddings, target_dim: int = 256):
    """Use Matryoshka embeddings or PCA to reduce dimensions."""
    # Matryoshka: truncate embeddings (works with bge, e5 models)
    return embeddings[:, :target_dim]

def use_scalar_quantization():
    """4x memory reduction with <1% accuracy loss."""
    # In Qdrant: quantization_config with scalar quantization
    # In Milvus: SCALAR quantization type
    pass

def use_binary_quantization():
    """32x memory reduction, ~5% accuracy loss."""
    # For massive scale where recall@10 is acceptable
    pass

def enable_disk_offload():
    """Store cold vectors on disk, hot in RAM."""
    # Qdrant: on_disk=True for original vectors
    # Milvus: tiered storage
    pass
Enter fullscreen mode Exit fullscreen mode

Decision Framework 

CHOOSING A VECTOR DATABASE:

START: Is this for production?
├─→ No (prototyping, local dev)
│   └─→ Use Chroma (simplest, Python-native)
└─→ Yes
    ├─→ Need fully managed, no ops?
    │   ├─→ Yes, AWS/GCP → Pinecone
    │   └─→ No, want control
    │       ├─→ Need GraphQL, hybrid search, multi-modal?
    │       │   └─→ Weaviate
    │       ├─→ Need GPU acceleration, massive scale (100M+)?
    │       │   └─→ Milvus (or Zilliz Cloud)
    │       └─→ General purpose, self-hosted, great filtering?
    │           └─→ Qdrant ⭐ RECOMMENDED
Enter fullscreen mode Exit fullscreen mode

Resources

Qdrant is the best all-rounder for self-hosted production. Pinecone for zero-ops. Weaviate for GraphQL/hybrid. Milvus for massive scale with GPU. Chroma for local dev.

Top comments (0)