17. ChromaDB Persistence

Chapter 17 of 18 · 20 min

KEY INSIGHT

ChromaDB persists automatically, but understanding the underlying storage format helps with backup and migration. ### Automatic Persistence When you use `PersistentClient`, ChromaDB saves to disk after each write operation: ```python import chromadb # This automatically persists to ./my_database client = chromadb.PersistentClient(path="./my_database") collection = client.create_collection("docs") # Add data collection.add(documents=["Test"], ids=["1"]) # Data is persisted immediately # On restart, data is available ``` ### Understanding the Storage Format ``` ./my_database/ â”œâ”€â”€ 062d4c40-1d54-4feb-bf7b-282f6c3e2e2a/ â”‚ â”œâ”€â”€ header.json â”‚ â”œâ”€â”€ data_level-0.db # SQLite database â”‚ â””â”€â”€ index/ â”‚ â””â”€â”€ 062d4c40-..._embedded.u8data # HNSW index â””â”€â”€ ... ``` The SQLite database stores documents and metadata. The index folder stores the HNSW vectors. ### Manual Backup ```python import shutil from datetime import datetime def backup_chroma(client_path: str, backup_path: str = None): """Create a timestamped backup of ChromaDB data.""" if backup_path is None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = f"./backup_{timestamp}" shutil.copytree(client_path, backup_path) print(f"Backup created at: {backup_path}") return backup_path # Create backup backup = backup_chroma("./my_database") ``` ### Restoration ```python def restore_chroma(backup_path: str, restore_path: str): """Restore ChromaDB from backup.""" import shutil import os # Remove existing database if present if os.path.exists(restore_path): shutil.rmtree(restore_path) shutil.copytree(backup_path, restore_path) return chromadb.PersistentClient(path=restore_path) # Restore from backup client = restore_chroma("./backup_20240115_143022", "./my_database") ``` ### Incremental Export For large databases, export incrementally: ```python def export_collection_to_json(collection, output_file: str, batch_size: int = 1000): """Export collection to JSON for migration or backup.""" import json total = collection.count() exported = 0 with open(output_file, 'w') as f: f.write('[\n') while exported < total: results = collection.get( include=["documents", "metadatas", "embeddings"], limit=batch_size, offset=exported ) for i in range(len(results['ids'])): record = { "id": results['ids'][i], "document": results['documents'][i], "metadata": results['metadatas'][i], "embedding": results['embeddings'][i] } json.dump(record, f) f.write(',\n' if exported + i + 1 < total else '\n') exported += len(results['ids']) print(f"Exported {exported}/{total}") f.write(']') # Export collection = client.get_collection("docs") export_collection_to_json(collection, "./export.json") ``` ### Import from Export ```python def import_collection_from_json(filepath: str, collection): """Import collection from JSON export.""" import json with open(filepath) as f: records = json.load(f) for i in range(0, len(records), 100): batch = records[i:i+100] collection.add( documents=[r["document"] for r in batch], embeddings=[r["embedding"] for r in batch], ids=[r["id"] for r in batch], metadatas=[r["metadata"] for r in batch] ) print(f"Imported {min(i+100, len(records))}/{len(records)}") ```

EXERCISE

Create a ChromaDB collection, add 100 documents, create a backup, delete the original, and restore from backup. Verify all documents are present.