Artificial Intelligence Database Migration Guide (for Linux + Bash)

Your data is ready for AI—your database probably isn’t. Whether you want semantic search, retrieval-augmented generation (RAG), or smarter recommendations, you need fast vector search and metadata filtering. This guide shows you how to migrate existing data into an AI-ready database on Linux using Bash-first tooling and open-source components.

You’ll get a reproducible, zero-downtime-friendly migration pipeline: export from a relational DB, clean and chunk text, generate embeddings, load into a vector database, and validate the results—all from your terminal.

Why this matters

• AI workloads need vectors, not just rows. Semantic search and RAG require embeddings (high-dimensional vectors) and specialized indexes.

• Ad hoc migrations are risky. Poor chunking, bad encodings, or mismatched vector dimensions lead to slow queries and poor relevance.

• Linux + Bash keeps it simple and auditable. Scripted steps are easy to automate, version, and roll back.

What you’ll build:

• A scripted pipeline to move documents from a relational source into a vector database (Qdrant), with clean chunking and embedding.

• A safe cutover strategy that supports blue/green collections and rollback.

Who this is for:

• Linux users comfortable with Bash, psql, curl, and Python.

• Teams migrating knowledge bases, tickets, docs, logs, or product catalogs to AI-ready search.

---

Prerequisites and installation

We’ll use:

• psql to export data from PostgreSQL (adaptable to other sources)

• Qdrant as the vector database (runs locally via Podman or Docker)

• Python + sentence-transformers to compute embeddings

Install core tools (choose your distro):

• Debian/Ubuntu (apt):

sudo apt-get update
sudo apt-get install -y curl jq git python3 python3-venv python3-pip podman postgresql-client

• Fedora/RHEL/CentOS (dnf):

sudo dnf install -y curl jq git python3 python3-virtualenv python3-pip podman postgresql

• openSUSE/SLE (zypper):

sudo zypper refresh
sudo zypper install -y curl jq git python3 python3-venv python3-pip podman postgresql

Start Qdrant (vector DB) locally:

mkdir -p ./qdrant_storage
podman run -d --name qdrant -p 6333:6333 \
  -v "$PWD/qdrant_storage:/qdrant/storage" \
  qdrant/qdrant:v1.7.4

Verify Qdrant is up:

curl -s http://localhost:6333/ | jq

Set up Python environment:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
pip install "sentence-transformers>=2.2" "qdrant-client>=1.7" "ujson>=5.8" "tqdm>=4.66"

Tip: First-time installs of sentence-transformers will download a CPU PyTorch wheel; allow a minute.

---

Step 1 — Assess and export your source data

Goal: Extract the text you want to search semantically (e.g., knowledge base articles).

Example: exporting from PostgreSQL into NDJSON (one JSON object per line).

Set a connection string:

export PGURI="postgresql://USER:PASSWORD@HOST:5432/DBNAME"

Quick profile:

psql "$PGURI" -Atc "SELECT count(*) FROM articles WHERE body IS NOT NULL;"
psql "$PGURI" -c "\d+ articles"

Export target fields to NDJSON:

psql "$PGURI" -c "\copy (
  SELECT row_to_json(t)
  FROM (
    SELECT id, title, body, updated_at
    FROM articles
    WHERE body IS NOT NULL
  ) t
) TO 'export.ndjson'"

Sanity check:

head -n 3 export.ndjson | jq
wc -l export.ndjson

Notes:

• Prefer NDJSON for streaming pipelines.

• If you don’t use PostgreSQL, export from your DB to NDJSON/CSV and adapt the next steps.

---

Step 2 — Clean and chunk text

Chunking is critical. Too-small chunks hurt recall; too-large chunks hurt precision and embedding quality. Start simple: 700–1,000 tokens per chunk with overlap.

Create a small prep script:

cat > prepare_chunks.py << 'PY'
import json, re, sys, hashlib
from datetime import datetime

def sanitize_text(s: str) -> str:
    if s is None:
        return ""
    s = re.sub(r"[\x00-\x1f\x7f]", " ", s)           # remove control chars
    s = re.sub(r"\s+", " ", s).strip()               # normalize whitespace
    # light PII masking example (extend as needed)
    s = re.sub(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", "[redacted_email]", s)
    return s

def chunk_text(text: str, max_words=250, overlap_words=40):
    # quick, token-agnostic chunking by words
    words = text.split()
    if not words:
        return []
    chunks = []
    start = 0
    while start < len(words):
        end = min(start + max_words, len(words))
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        if end == len(words):
            break
        start = max(0, end - overlap_words)
    return chunks

in_path  = sys.argv[1] if len(sys.argv) > 1 else "export.ndjson"
out_path = sys.argv[2] if len(sys.argv) > 2 else "chunks.ndjson"

model_name = "sentence-transformers/all-MiniLM-L6-v2"
chunk_method = "words250_overlap40"

outf = open(out_path, "w", encoding="utf-8")
with open(in_path, "r", encoding="utf-8") as f:
    for line in f:
        if not line.strip():
            continue
        doc = json.loads(line)
        title = sanitize_text(doc.get("title"))
        body  = sanitize_text(doc.get("body"))
        text  = (title + "\n\n" + body).strip()
        chunks = chunk_text(text, max_words=250, overlap_words=40)

        base_id = str(doc.get("id"))
        for i, ch in enumerate(chunks):
            cid = f"{base_id}::chunk::{i}"
            payload = {
                "source_id": base_id,
                "chunk_index": i,
                "title": title[:512],
                "updated_at": doc.get("updated_at"),
                "model": model_name,
                "chunk_method": chunk_method,
                "length": len(ch),
            }
            # store the raw text to embed later
            rec = {"id": cid, "text": ch, "payload": payload}
            outf.write(json.dumps(rec, ensure_ascii=False) + "\n")
outf.close()
print(f"Wrote chunks to {out_path}")
PY

Run it:

python prepare_chunks.py export.ndjson chunks.ndjson
head -n 2 chunks.ndjson | jq

---

Step 3 — Embed and load into a vector database (Qdrant)

We’ll embed each chunk using a small, fast model and upsert into Qdrant. The chosen model:

• all-MiniLM-L6-v2 (384 dimensions, cosine distance)

Create the loader:

cat > embed_and_load.py << 'PY'
import json, sys
from pathlib import Path
from tqdm import tqdm

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.http import models as qm

chunks_path = sys.argv[1] if len(sys.argv) > 1 else "chunks.ndjson"
collection = sys.argv[2] if len(sys.argv) > 2 else "docs_v1"

# 1) Connect to Qdrant
client = QdrantClient(url="http://localhost:6333")

# 2) Load embedding model
model_name = "sentence-transformers/all-MiniLM-L6-v2"
model = SentenceTransformer(model_name)
dim = model.get_sentence_embedding_dimension()

# 3) Create collection if not exists
existing = [c.name for c in client.get_collections().collections]
if collection not in existing:
    client.create_collection(
        collection_name=collection,
        vectors_config=qm.VectorParams(size=dim, distance=qm.Distance.COSINE),
    )
    print(f"Created collection {collection} with dim={dim}")

# 4) Stream chunks, batch-embed, and upsert
BATCH = 256
buf_ids, buf_vecs, buf_payloads = [], [], []

def flush():
    if not buf_ids:
        return
    client.upsert(
        collection_name=collection,
        points=qm.Batch(
            ids=buf_ids,
            vectors=buf_vecs,
            payloads=buf_payloads
        )
    )
    buf_ids.clear(); buf_vecs.clear(); buf_payloads.clear()

lines = 0
with open(chunks_path, "r", encoding="utf-8") as f:
    texts, records = [], []
    for line in f:
        if not line.strip():
            continue
        rec = json.loads(line)
        texts.append(rec["text"])
        records.append(rec)
        lines += 1

        if len(texts) >= BATCH:
            vecs = model.encode(texts, normalize_embeddings=True).tolist()
            for v, r in zip(vecs, records):
                buf_ids.append(r["id"])
                buf_vecs.append(v)
                buf_payloads.append(r["payload"] | {"text_preview": r["text"][:200]})
            flush()
            texts, records = [], []

    # final partial
    if texts:
        vecs = model.encode(texts, normalize_embeddings=True).tolist()
        for v, r in zip(vecs, records):
            buf_ids.append(r["id"])
            buf_vecs.append(v)
            buf_payloads.append(r["payload"] | {"text_preview": r["text"][:200]})
        flush()

print(f"Upserted {lines} chunks into {collection}")
PY

Run it:

python embed_and_load.py chunks.ndjson docs_v1

Check collection stats:

curl -s localhost:6333/collections | jq
curl -s localhost:6333/collections/docs_v1 | jq

---

Step 4 — Validate semantic search end-to-end

Generate a query embedding and search Qdrant.

Compute a query vector:

VEC=$(python - <<'PY'
from sentence_transformers import SentenceTransformer
m = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
v = m.encode(["How do I reset my account password?"], normalize_embeddings=True)[0].tolist()
import json; print(json.dumps(v))
PY
)

Search Qdrant:

curl -s -X POST localhost:6333/collections/docs_v1/points/search \
  -H 'Content-Type: application/json' \
  -d "{
        \"vector\": $VEC,
        \"limit\": 5,
        \"with_payload\": true
      }" | jq '.result[] | {id, score, title:.payload.title, preview:.payload.text_preview}'

Add a metadata filter example (e.g., updated after a date):

curl -s -X POST localhost:6333/collections/docs_v1/points/search \
  -H 'Content-Type: application/json' \
  -d "{
        \"vector\": $VEC,
        \"limit\": 5,
        \"with_payload\": true,
        \"filter\": {\"must\": [{\"key\":\"updated_at\",\"range\":{\"gte\": \"2024-01-01\"}}]}
      }" | jq '.result[] | {id, score, title:.payload.title}'

If results look irrelevant:

• Revisit chunk size/overlap in prepare_chunks.py

• Try a stronger model (e.g., BAAI/bge-small-en-v1.5) and rebuild a new collection (docs_v2)

• Verify text cleaning preserves key entities and context

---

Step 5 — Plan cutover with confidence

Avoid risky “big switch” moments. Use a blue/green index strategy:

• Build in parallel:

  • Keep current search live.
  • Create a new collection (e.g., docs_v2) with refined chunking/model.

• Shadow test:

  • Mirror a subset of queries to both collections.
  • Compare top-k overlap and qualitative relevance offline.

• Dual write:

  • As new/updated documents arrive, update both collections.

• Cutover:

  • Toggle an environment variable or config flag to query docs_v2.
  • Keep docs_v1 for rollback until confidence is high.

• Clean up:

  • Decommission old collection after a stability window.

Example commands:

# new collection name
python embed_and_load.py chunks.ndjson docs_v2

# monitor sizes
curl -s localhost:6333/collections | jq '.collections[] | {name:.name}'

# backup vector DB storage (since we used a host volume)
tar czf qdrant_storage_backup.tgz qdrant_storage/

---

Real-world example: Migrating a knowledge base

• Source: articles(id, title, body, updated_at) in PostgreSQL

• Target: Qdrant docs_v1 with all-MiniLM-L6-v2 (384-dim cosine)

• Steps: 1) Export to NDJSON with row_to_json to preserve structural fidelity. 2) Clean text, mask emails, chunk by ~250 words with 40-word overlap. 3) Embed and upsert in batches of 256; keep a short text preview in payload for debugging. 4) Validate top-5 results for common support queries; filter by updated_at during incidents. 5) Iterate chunking/model, build docs_v2, shadow test, and cut over.

Outcome: Faster, more relevant search for support teams, and a repeatable pipeline for future updates.

---

Troubleshooting tips

• Embedding dimension mismatch: Ensure the collection’s vector size matches the model’s embedding size. We derive it programmatically in the loader.

• Slow performance: Increase batch size, ensure normalize_embeddings=True for cosine, and pin to a small model first.

• Encoding issues: Use UTF-8 and sanitize control characters. Validate with jq and iconv if needed.

• Storage growth: Keep only a short text_preview in payload; store full text elsewhere if compliance requires.

---

Conclusion and next steps

You now have a Linux-native, scriptable path to migrate data into an AI-ready database:

• Export from your source DB

• Clean and chunk text

• Embed with a solid baseline model

• Load into a vector database

• Validate, iterate, and cut over safely

Next steps:

• Parameterize your pipeline (env vars for collection/model names).

• Add CI/CD to rebuild indexes on schema/model changes.

• Introduce hybrid search (BM25 + vectors) if your use case needs keyword precision.

• Layer on RAG or recommendations using your new vector index.

If this was helpful, try running the pipeline on a small subset of your production data today—and measure how semantic search improves your users’ outcomes. Then iterate and scale with confidence.