Capsule 07: Zero-Downtime Migration (ChromaDB → Pinecone)

Capsule description

Migrating a vector database in production requires protecting the end user. This capsule walks you step by step through a gradual pattern to move data and traffic from ChromaDB to Pinecone without noticeable interruptions. We cover ID mapping, metadata schemas, the dual-write strategy, canary routing, and post-migration validation with complete Python scripts.

Why migrate from ChromaDB to Pinecone?

ChromaDB is excellent for development and prototypes. But when your RAG grows — more than ~500K vectors, a need for an SLA, multi-region, or simply not wanting to maintain infra — Pinecone offers zero-ops, automatic scaling, and low latency. The migration does not have to be a "big bang": with this capsule's pattern you do it without downtime.

Critical differences: ChromaDB vs Pinecone

Before migrating, understand the differences that impact the migration design:

These differences determine the ID mapping, the metadata transformation, and whether you re-embed or export existing vectors.

Recommended strategy: the 5 phases

Overview

Phase 1: Preparation
   └─ ID mapping, metadata, re-embed vs export decision
Phase 2: Initial load into the target
   └─ Batch migration script
Phase 3: Temporary dual-write
   └─ Write to both for N days
Phase 4: Dual-read and validation
   └─ Compare results, accuracy, latency
Phase 5: Canary + cutover
   └─ Redirect % of traffic, monitor, cut off the source

Phase 1: Preparation

1.1 ID mapping (ChromaDB UUIDs → Pinecone strings)

ChromaDB uses internal UUIDs; Pinecone accepts arbitrary strings. You can:

Option A — Keep the UUID as a string (recommended):

# ChromaDB returns IDs as str "550e8400-e29b-41d4-a716-446655440000"
# Pinecone accepts up to 512 bytes — a UUID fits perfectly
pinecone_id = chroma_id  # no changes

Option B — Prefix to avoid collisions:

If at some point you mixed IDs from different collections or systems:

def chroma_to_pinecone_id(chroma_id: str, prefix: str = "doc_") -> str:
    """Map a ChromaDB ID to a Pinecone ID with a prefix."""
    return f"{prefix}{chroma_id.replace('-', '')}" if prefix else chroma_id

Option C — Semantic IDs (if you have them):

If your documents have a business ID (e.g. doc_12345), use it directly in both systems to make traceability easier.

# If you stored metadata with "doc_id" in ChromaDB:
metadata = chroma_result["metadatas"][0]
pinecone_id = metadata.get("doc_id") or chroma_result["ids"][0]

1.2 Metadata mapping

Pinecone only accepts scalar types. ChromaDB is more permissive. You need a transformation function:

from typing import Any

def chroma_metadata_to_pinecone(metadata: dict[str, Any]) -> dict[str, str | int | float | bool]:
    """Convert ChromaDB metadata to a Pinecone-compatible schema."""
    allowed = (str, int, float, bool)
    result = {}
    for k, v in metadata.items():
        if v is None:
            continue
        if isinstance(v, allowed):
            result[k] = v
        elif isinstance(v, list) and all(isinstance(x, str) for x in v):
            result[k] = v  # list[str] allowed in Pinecone
        else:
            result[k] = str(v)  # dict, list[dict], datetime, etc. → serialize
    return result

Typical rules:

• None → omit
• datetime → ISO string
• list[dict] → JSON string
• Key names: avoid special characters; Pinecone uses . for nested (e.g. user.name)

1.3 Existing embeddings: re-embed or export?

Export (recommended if the model does not change):

# ChromaDB stores vectors; you can read them without re-embedding
collection = client.get_collection("my_collection")
results = collection.get(include=["embeddings", "metadatas", "documents"])
# results["embeddings"] is list[list[float]]

Re-embed: Use it if you migrate to another model (e.g. text-embedding-3-large) or if ChromaDB does not give you easy batch access to vectors.

Phase 2: Complete batch migration script

Complete script to migrate a ChromaDB collection to a Pinecone index:

#!/usr/bin/env python3
"""
ChromaDB → Pinecone migration (initial batch).
Usage: python migrate_chroma_to_pinecone.py --chroma-path ./chroma_db --collection docs
"""
import argparse
import chromadb
from pinecone import Pinecone
from chroma_metadata_to_pinecone import chroma_metadata_to_pinecone  # previous function


def migrate_collection(
    chroma_path: str,
    collection_name: str,
    pinecone_api_key: str,
    pinecone_index: str,
    pinecone_namespace: str = "default",
    batch_size: int = 100,
) -> dict:
    """
    Migrate a ChromaDB collection to Pinecone.
    Returns stats: {uploaded, failed, errors}.
    """
    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_collection(collection_name)
    pc = Pinecone(api_key=pinecone_api_key)
    index = pc.Index(pinecone_index)

    # Get all data (embeddings included)
    results = collection.get(
        include=["embeddings", "metadatas", "documents"]
    )
    ids = results["ids"]
    embeddings = results["embeddings"]
    metadatas = results.get("metadatas") or [{}] * len(ids)
    documents = results.get("documents") or [None] * len(ids)

    uploaded = 0
    failed = 0
    errors = []

    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i : i + batch_size]
        batch_embeddings = embeddings[i : i + batch_size]
        batch_metadatas = metadatas[i : i + batch_size]
        batch_docs = documents[i : i + batch_size]

        vectors = []
        for j, (cid, emb, meta, doc) in enumerate(
            zip(batch_ids, batch_embeddings, batch_metadatas, batch_docs)
        ):
            meta_clean = chroma_metadata_to_pinecone(meta or {})
            if doc is not None:
                meta_clean["text"] = doc[:40_000]  # Pinecone metadata limit ~40KB
            vectors.append({
                "id": cid,
                "values": emb,
                "metadata": meta_clean,
            })

        try:
            index.upsert(
                vectors=vectors,
                namespace=pinecone_namespace,
            )
            uploaded += len(vectors)
        except Exception as e:
            failed += len(vectors)
            errors.append({"batch": i // batch_size, "error": str(e)})

    return {"uploaded": uploaded, "failed": failed, "errors": errors}


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", required=True)
    parser.add_argument("--collection", required=True)
    parser.add_argument("--pinecone-index", required=True)
    parser.add_argument("--namespace", default="default")
    parser.add_argument("--batch-size", type=int, default=100)
    args = parser.parse_args()

    import os
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        raise SystemExit("PINECONE_API_KEY not set")

    stats = migrate_collection(
        chroma_path=args.chroma_path,
        collection_name=args.collection,
        pinecone_api_key=api_key,
        pinecone_index=args.pinecone_index,
        pinecone_namespace=args.namespace,
        batch_size=args.batch_size,
    )
    print(f"Migration: {stats['uploaded']} uploaded, {stats['failed']} failed")
    if stats["errors"]:
        for e in stats["errors"][:5]:
            print(f"  Error: {e}")

Phase 3: Dual-write during the transition

During the transition window (typically 7–14 days), you write to both systems. This way any new or updated document ends up in both ChromaDB and Pinecone.

Implementation with an abstraction

from abc import ABC, abstractmethod
from typing import Optional
import chromadb
from pinecone import Pinecone


class VectorStore(ABC):
    @abstractmethod
    def upsert(self, ids: list[str], embeddings: list[list[float]], metadatas: list[dict]):
        pass

    @abstractmethod
    def query(self, embedding: list[float], top_k: int = 5, filter_: Optional[dict] = None):
        pass


class ChromaStore(VectorStore):
    def __init__(self, path: str, collection: str):
        self.client = chromadb.PersistentClient(path=path)
        self.collection = self.client.get_collection(collection)

    def upsert(self, ids, embeddings, metadatas):
        self.collection.upsert(ids=ids, embeddings=embeddings, metadatas=metadatas)

    def query(self, embedding, top_k=5, filter_=None):
        return self.collection.query(
            query_embeddings=[embedding],
            n_results=top_k,
            where=filter_,
            include=["metadatas", "distances"],
        )


class PineconeStore(VectorStore):
    def __init__(self, api_key: str, index: str, namespace: str = "default"):
        pc = Pinecone(api_key=api_key)
        self.index = pc.Index(index)
        self.namespace = namespace

    def upsert(self, ids, embeddings, metadatas):
        vectors = [
            {"id": iid, "values": emb, "metadata": chroma_metadata_to_pinecone(m or {})}
            for iid, emb, m in zip(ids, embeddings, metadatas)
        ]
        self.index.upsert(vectors=vectors, namespace=self.namespace)

    def query(self, embedding, top_k=5, filter_=None):
        r = self.index.query(
            vector=embedding,
            top_k=top_k,
            filter=filter_,
            namespace=self.namespace,
            include_metadata=True,
        )
        return {"ids": [m.id for m in r.matches], "metadatas": [m.metadata for m in r.matches]}


class DualWriteStore(VectorStore):
    """Writes to both, reads from the source (ChromaDB) during dual-write."""

    def __init__(self, chroma: ChromaStore, pinecone: PineconeStore):
        self.chroma = chroma
        self.pinecone = pinecone

    def upsert(self, ids, embeddings, metadatas):
        self.chroma.upsert(ids, embeddings, metadatas)
        try:
            self.pinecone.upsert(ids, embeddings, metadatas)
        except Exception as e:
            # Log but don't fail — ChromaDB is the source of truth
            import logging
            logging.warning(f"Dual-write Pinecone failed: {e}")

    def query(self, embedding, top_k=5, filter_=None):
        return self.chroma.query(embedding, top_k, filter_)

During dual-write, the reader is still ChromaDB. Only when you validate equivalence do you move to dual-read and then to pure Pinecone.

Phase 4: Dual-read and equivalence validation

4.1 Result comparison

You need a set of real queries (e.g. 50–200 production or synthetic questions). For each query:

1. You generate the embedding.
2. You query ChromaDB and Pinecone.
3. You compare the top-k returned (IDs, scores, order).

def compare_results(chroma_res: dict, pinecone_res: dict, top_k: int = 5) -> dict:
    """
    Compare ChromaDB vs Pinecone results.
    Returns: overlap (Jaccard), order_correlation, score_diff.
    """
    chroma_ids = set(chroma_res.get("ids", [[]])[0][:top_k])
    pinecone_ids = set(pinecone_res.get("ids", [])[:top_k])

    overlap = len(chroma_ids & pinecone_ids) / top_k if top_k else 0
    jaccard = len(chroma_ids & pinecone_ids) / len(chroma_ids | pinecone_ids) if chroma_ids or pinecone_ids else 1.0

    # Score difference (if both use cosine)
    chroma_dists = chroma_res.get("distances", [[]])[0][:top_k]
    pinecone_scores = [m.get("score") for m in (pinecone_res.get("metadatas") or [])[:top_k]]
    score_diff = 0
    if chroma_dists and pinecone_scores:
        score_diff = sum(abs(c - p) for c, p in zip(chroma_dists, pinecone_scores)) / min(len(chroma_dists), len(pinecone_scores))

    return {
        "overlap": overlap,
        "jaccard": jaccard,
        "score_diff": score_diff,
    }

4.2 Complete validation script

def validate_migration(
    chroma: ChromaStore,
    pinecone: PineconeStore,
    test_queries: list[str],
    embedding_fn,
    top_k: int = 5,
) -> dict:
    """
    Validate that ChromaDB and Pinecone return equivalent results.
    Returns aggregate metrics.
    """
    overlaps = []
    score_diffs = []

    for q in test_queries:
        emb = embedding_fn(q)
        chroma_res = chroma.query(emb, top_k=top_k)
        pinecone_res = pinecone.query(emb, top_k=top_k)
        cmp = compare_results(chroma_res, pinecone_res, top_k)
        overlaps.append(cmp["overlap"])
        score_diffs.append(cmp["score_diff"])

    return {
        "mean_overlap": sum(overlaps) / len(overlaps) if overlaps else 0,
        "min_overlap": min(overlaps) if overlaps else 0,
        "mean_score_diff": sum(score_diffs) / len(score_diffs) if score_diffs else 0,
        "n_queries": len(test_queries),
    }

Typical criteria to advance:

• mean_overlap >= 0.9 (90% of the top-k IDs match)
• mean_score_diff < 0.05 (very similar scores)
• If they are not met, investigate: different filters, normalization, insertion order.

Phase 5: Canary and cutover

5.1 Canary traffic routing

Instead of cutting all traffic at once, you redirect a percentage (5% → 25% → 50% → 100%) to Pinecone over time windows (e.g. 24–48 h per step).

import random

def get_vector_store(use_pinecone_prob: float = 0) -> VectorStore:
    """Return Chroma or Pinecone based on a probability (canary)."""
    chroma = ChromaStore(path="./chroma_db", collection="docs")
    pinecone = PineconeStore(api_key="...", index="docs", namespace="default")
    if random.random() < use_pinecone_prob:
        return pinecone
    return chroma

# In your API: store = get_vector_store(use_pinecone_prob=0.05)  # 5% canary

5.2 Metrics to decide advance or rollback

Rule: If two of three metrics get worse, stop the advance and keep dual-write. Don't increase the canary % until you resolve it.

5.3 Pre-cutover checklist

• Target dataset synced and validated
• Dual-write active for a minimum window (e.g. 7 days)
• Dual-read validated (mean_overlap ≥ 0.9)
• Canary plan defined (percentage and duration per step)
• Rollback tested in staging
• Rollback criteria documented and known by the team

Rollback criteria

Trigger a rollback (return to ChromaDB at 100%) if:

1. Accuracy drops below the threshold (e.g. mean_overlap < 0.85)
2. Error rate exceeds the limit (e.g. > 0.5%)
3. Latency gets sustainedly worse (p95 > 1.5× the source)
4. Critical incidents related to Pinecone (outages, timeouts)

Troubleshooting

1. "Dual-write generates inconsistencies between Chroma and Pinecone"

Typical causes: Different write order, silent failures in one of the two, metadata that doesn't transform well.

Solution:

• Validate that the write order is idempotent: Chroma first, Pinecone second.
• If Pinecone fails, log the error and retry in an async queue; don't leave documents only in Chroma.
• Review chroma_metadata_to_pinecone: disallowed types (dict, list of objects) cause errors in Pinecone.

2. "The canary shows intermittent latency or spikes"

Causes: Cold starts (serverless Pinecone), network, load differences between moments.

Solution:

• Increase the observation window (e.g. 48 h minimum per step).
• Compare p50, p95, p99; if p50 is stable and p99 is high, it may be a cold start.
• Consider warm-up: run periodic low-volume queries to keep the index "warm".

3. "ChromaDB IDs don't match the expected Pinecone IDs"

Causes: UUID with a different format, encoding, or IDs generated automatically by ChromaDB that you didn't store.

Solution:

• Use include=["embeddings","metadatas","documents"] in get() and reconstruct a doc_id → chroma_uuid mapping if you stored doc_id in metadata.
• Otherwise, the ChromaDB UUID is valid in Pinecone; pass it as-is. Verify that you are not adding/removing hyphens or prefixes by mistake.

4. "Metadata with complex types fails to insert into Pinecone"

Causes: Pinecone only accepts str, int, float, bool, list[str].

Solution:

• Serialize to a string: datetime → ISO, dict/list[dict] → json.dumps().
• Avoid keys with . if you don't want nested; normalize names (snake_case recommended).
• Validate with a dry-run script that attempts to upsert a sample before the full migration.

5. "The team pushes to cut over quickly"

Risk: Without clear rollback criteria and without enough validation, you increase the probability of incidents.

Solution:

• Document rollback criteria (accuracy, error rate, latency) and share them with the team.
• Define a "minimum viable" dual-write (e.g. 7 days) and canary (e.g. 5% → 25% → 100% in 3 steps).
• If there is a rush, shorten the window but don't remove the validation or the rollback criteria.

Exercises

Exercise 1: Metadata mapping

You have metadata in ChromaDB with {"created_at": datetime(2025,3,1), "tags": ["a","b"], "nested": {"x": 1}}. Write the Pinecone-compatible version.

from datetime import datetime
import json

metadata = {
    "created_at": datetime(2025, 3, 1),
    "tags": ["a", "b"],
    "nested": {"x": 1},
}

def to_pinecone(m):
    out = {}
    for k, v in m.items():
        if v is None:
            continue
        if isinstance(v, (str, int, float, bool)):
            out[k] = v
        elif isinstance(v, list) and all(isinstance(x, str) for x in v):
            out[k] = v
        elif isinstance(v, datetime):
            out[k] = v.isoformat()
        elif isinstance(v, (dict, list)):
            out[k] = json.dumps(v)
    return out

print(to_pinecone(metadata))
# {"created_at": "2025-03-01T00:00:00", "tags": ["a","b"], "nested": "{\"x\": 1}"}

Exercise 2: Result comparison script

Implement a function that, given two ordered lists of IDs (Chroma vs Pinecone), returns the positional overlap (how many IDs match at the same position) and the set overlap (how many IDs are in both top-k).

def overlap_metrics(chroma_ids: list[str], pinecone_ids: list[str], k: int = 5) -> dict:
    chroma_ids = chroma_ids[:k]
    pinecone_ids = pinecone_ids[:k]
    set_c = set(chroma_ids)
    set_p = set(pinecone_ids)
    position_overlap = sum(1 for i in range(min(len(chroma_ids), len(pinecone_ids)))
                          if chroma_ids[i] == pinecone_ids[i])
    set_overlap = len(set_c & set_p) / k if k else 0
    jaccard = len(set_c & set_p) / len(set_c | set_p) if set_c or set_p else 1.0
    return {
        "position_overlap": position_overlap,
        "position_overlap_ratio": position_overlap / k if k else 0,
        "set_overlap": set_overlap,
        "jaccard": jaccard,
    }

Exercise 3: Dual-write with retry

Extend DualWriteStore so that, if the upsert to Pinecone fails, it retries up to 3 times with exponential backoff (1s, 2s, 4s) before logging the failure.

import time

def upsert_with_retry(pinecone_store: PineconeStore, ids, embeddings, metadatas, max_retries=3):
    for attempt in range(max_retries):
        try:
            pinecone_store.upsert(ids, embeddings, metadatas)
            return
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

# In DualWriteStore.upsert:
try:
    upsert_with_retry(self.pinecone, ids, embeddings, metadatas)
except Exception as e:
    logging.warning(f"Dual-write Pinecone failed after retries: {e}")

Exercise 4: Canary by header

Instead of a random probability, implement a canary by HTTP header: if X-Use-Pinecone: true, use Pinecone; otherwise, ChromaDB. What advantage does it have over the random approach?

def get_store_from_header(header_value: str | None) -> VectorStore:
    use_pinecone = (header_value or "").lower() in ("1", "true", "yes")
    return pinecone if use_pinecone else chroma

# Advantage: It allows manual testing (curl -H "X-Use-Pinecone: true" ...)
# and controlled A/B testing (only certain users/tenant via header).
# Random is better for a "blind" canary of a traffic percentage.

Exercise 5: Count validation

Before comparing results per query, validate that the total number of vectors in ChromaDB matches that of Pinecone. Write a script that returns both counts.

def count_chroma(client, collection_name: str) -> int:
    col = client.get_collection(collection_name)
    return col.count()

def count_pinecone(index, namespace: str = "default") -> int:
    # Pinecone: describe_index_stats() returns the count per namespace
    stats = index.describe_index_stats()
    ns = stats.namespaces.get(namespace)
    return ns.vector_count if ns else 0

# Usage:
chroma_count = count_chroma(client, "docs")
pinecone_count = count_pinecone(pc_index, "default")
assert chroma_count == pinecone_count, f"Mismatch: {chroma_count} vs {pinecone_count}"

Exercise 6: Re-embed vs export decision

You have 100K documents in ChromaDB with text-embedding-ada-002 embeddings. You want to migrate to Pinecone. Options: (A) export vectors, (B) re-embed with the same model, (C) re-embed with text-embedding-3-small. Draw up a pros/cons table and recommend one option based on: (i) limited time, (ii) you want to improve retrieval quality.

Summary

• Zero-downtime migration requires phases: preparation (ID/metadata mapping, re-embed decision), batch load, dual-write, dual-read validation, canary, and cutover.
• ChromaDB vs Pinecone differ in IDs (UUID vs string), metadata (allowed types), and filters; explicit mapping avoids errors.
• Dual-write keeps both systems in sync; the reader remains ChromaDB until you validate equivalence.
• Validation is based on result overlap (mean_overlap ≥ 0.9) and comparison of latencies/error rate.
• Canary reduces risk by redirecting a % of traffic to Pinecone; if two of three metrics get worse, stop the advance.
• Rollback must be defined (accuracy, error rate, latency) and tested in staging before the cutover.
• With this pattern you are ready to consolidate the final production-readiness checklist.

Additional resources

1. Martin Fowler — Canary Release — Canary pattern applied to releases
2. Pinecone — Migration guide — Official migration guide
3. ChromaDB — Exporting data — How to export collections
4. Pinecone — Metadata filtering — Filter syntax
5. Strangler Fig Pattern — Pattern to replace systems gradually
6. Zero-downtime migrations (blog) — Downtime-free deployment strategies
7. Pinecone — Python SDK — Official client
8. ChromaDB — Python Client API — API reference

Estimated time: 45–60 minutes
Next: 08-project-production-readiness-checklist.md