Using Elasticsearch as a Haystack document store

This elasticsearch haystack document store tutorial walks you through wiring Elasticsearch into a Haystack pipeline from a clean environment to a production-ready retriever. You’ll provision the store, configure dense and sparse retrieval, index documents with metadata filtering, and add the observability hooks you need before shipping to prod.

Step 1: Provision Elasticsearch with the right settings

Haystack’s ElasticsearchDocumentStore expects a running cluster with at least one index. For local development, Docker Compose is the fastest path. For staging and prod, use a managed service (Elastic Cloud, AWS OpenSearch, or a self-managed cluster with dedicated master nodes).

Create docker-compose.yml:

version: "3.8"
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms1g -Xmx1g"
    ports:
      - "9200:9200"
      - "9300:9300"
    volumes:
      - esdata:/usr/share/elasticsearch/data
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:9200/_cluster/health || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 10

volumes:
  esdata:

Start it:

docker compose up -d

Verify the cluster is green:

curl -s http://localhost:9200/_cluster/health | jq .status
# expected: "green" or "yellow" (single node)

Why these settings matter: Disabling security speeds up local iteration. In prod, enable TLS and create a dedicated service account with manage_index, read, write, and monitor privileges on the target index pattern. The JVM heap at 1 GB is a minimum; size it to 50% of available RAM up to 32 GB.

Step 2: Install Haystack and the Elasticsearch integration

Use a virtual environment. Haystack 2.x splits integrations into separate packages.

python -m venv .venv && source .venv/bin/activate
pip install --upgrade pip
pip install "haystack-ai>=2.0" "elasticsearch-haystack>=2.0" sentence-transformers

The elasticsearch-haystack package contains the document store implementation. sentence-transformers provides local embedding models for dense retrieval.

Verify imports work:

from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore
from haystack import Document
print("Imports OK")

Step 3: Initialize the document store with a schema that supports hybrid search

Haystack maps each Document to an Elasticsearch document. You control the index mapping through the index parameter and optional custom_mapping. For hybrid retrieval (BM25 + dense vectors), define both a text field and a dense_vector field.

from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore

document_store = ElasticsearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_docs",
    embedding_dim=384,  # matches all-MiniLM-L6-v2
    similarity="cosine",
    # Custom mapping adds a keyword field for exact metadata filtering
    custom_mapping={
        "properties": {
            "meta": {
                "type": "object",
                "dynamic": True,
                "properties": {
                    "source_id": {"type": "keyword"},
                    "category": {"type": "keyword"},
                    "timestamp": {"type": "date"}
                }
            }
        }
    }
)

Verification: The store creates the index on first write. Check it exists:

curl -s http://localhost:9200/haystack_docs/_mapping | jq .

You should see text (text), embedding (dense_vector, 384 dims, cosine), and meta (object with your keyword sub-fields).

Step 4: Generate embeddings and write documents

Haystack’s Document class carries content, metadata, and optionally a precomputed embedding. For a real pipeline, compute embeddings in a separate step (batch, GPU, caching). Here we do it inline for clarity.

from haystack import Document
from sentence_transformers import SentenceTransformer

embedder = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

raw_docs = [
    {"text": "Haystack 2.x uses a component-based pipeline architecture.", "source_id": "doc-1", "category": "architecture", "timestamp": "2024-01-15"},
    {"text": "ElasticsearchDocumentStore supports BM25 and dense vector search.", "source_id": "doc-2", "category": "retrieval", "timestamp": "2024-01-16"},
    {"text": "Hybrid retrieval combines sparse and dense scores with reciprocal rank fusion.", "source_id": "doc-3", "category": "retrieval", "timestamp": "2024-01-17"},
    {"text": "Metadata filtering uses Elasticsearch keyword fields for exact matches.", "source_id": "doc-4", "category": "filtering", "timestamp": "2024-01-18"},
]

documents = []
for d in raw_docs:
    emb = embedder.encode(d["text"]).tolist()
    documents.append(
        Document(
            content=d["text"],
            meta={"source_id": d["source_id"], "category": d["category"], "timestamp": d["timestamp"]},
            embedding=emb
        )
    )

document_store.write_documents(documents, policy="overwrite")
print(f"Indexed {document_store.count_documents()} documents")

Verification: Confirm the count and inspect a document:

curl -s "http://localhost:9200/haystack_docs/_search?size=1&pretty" | jq '.hits.hits[0]._source'

You should see content, meta, and embedding (array of 384 floats).

Step 5: Configure retrievers for sparse, dense, and hybrid search

Haystack provides three retriever classes for Elasticsearch:

• ElasticsearchBM25Retriever — classic lexical search
• ElasticsearchEmbeddingRetriever — dense vector search
• ElasticsearchHybridRetriever — runs both and fuses with reciprocal rank fusion (RRF)

from haystack_integrations.components.retrievers.elasticsearch import (
    ElasticsearchBM25Retriever,
    ElasticsearchEmbeddingRetriever,
    ElasticsearchHybridRetriever,
)

bm25_retriever = ElasticsearchBM25Retriever(document_store=document_store, top_k=5)
embedding_retriever = ElasticsearchEmbeddingRetriever(document_store=document_store, top_k=5)
hybrid_retriever = ElasticsearchHybridRetriever(
    document_store=document_store,
    top_k=5,
    # RRF constant; 60 is the Elasticsearch default
    rrf_constant=60
)

Run a query through each retriever

query = "How does hybrid retrieval work in Haystack?"
query_embedding = embedder.encode(query).tolist()

# BM25 only
bm25_results = bm25_retriever.run(query=query, top_k=3)
print("BM25:", [d.content[:60] for d in bm25_results["documents"]])

# Dense only
emb_results = embedding_retriever.run(query_embedding=query_embedding, top_k=3)
print("Dense:", [d.content[:60] for d in emb_results["documents"]])

# Hybrid (RRF)
hybrid_results = hybrid_retriever.run(query=query, query_embedding=query_embedding, top_k=3)
print("Hybrid:", [d.content[:60] for d in hybrid_results["documents"]])

Expected behavior: BM25 surfaces keyword matches (“hybrid”, “retrieval”). Dense surfaces semantic matches. Hybrid blends them — typically the best recall for RAG.

Step 6: Add metadata filtering to retriever calls

Filters apply to the meta object. Use Elasticsearch’s query DSL via Haystack’s filters parameter (a dict that maps to a bool filter clause).

# Only retrieval category, after a certain date
filters = {
    "operator": "AND",
    "conditions": [
        {"field": "meta.category", "operator": "==", "value": "retrieval"},
        {"field": "meta.timestamp", "operator": ">=", "value": "2024-01-16"}
    ]
}

filtered = hybrid_retriever.run(
    query=query,
    query_embedding=query_embedding,
    top_k=5,
    filters=filters
)
print("Filtered:", [f"{d.meta['category']} | {d.content[:50]}" for d in filtered["documents"]])

Verification: Only doc-2 and doc-3 should return. The filter pushes down to Elasticsearch — no post-filtering in Python.

Step 7: Build a complete indexing pipeline for production

Real systems ingest continuously. Haystack’s Pipeline class lets you chain components: file conversion, cleaning, splitting, embedding, writing. Here’s a minimal production-grade indexing pipeline.

from haystack import Pipeline
from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter

indexing_pipeline = Pipeline()
indexing_pipeline.add_component("converter", PyPDFToDocument())
indexing_pipeline.add_component("cleaner", DocumentCleaner(remove_empty_lines=True, remove_extra_whitespaces=True))
indexing_pipeline.add_component("splitter", DocumentSplitter(split_by="sentence", split_length=5, split_overlap=1))
indexing_pipeline.add_component("embedder", SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
indexing_pipeline.add_component("writer", DocumentWriter(document_store=document_store, policy="overwrite"))

indexing_pipeline.connect("converter", "cleaner")
indexing_pipeline.connect("cleaner", "splitter")
indexing_pipeline.connect("splitter", "embedder")
indexing_pipeline.connect("embedder", "writer")

# Run on a directory of PDFs
import glob
pdf_paths = glob.glob("data/*.pdf")
results = indexing_pipeline.run({"converter": {"sources": pdf_paths}})
print(f"Indexed {results['writer']['documents_written']} documents from {len(pdf_paths)} PDFs")

Key production notes:

• DocumentSplitter with sentence-aware splitting preserves semantic boundaries better than character splitting.
• SentenceTransformersDocumentEmbedder batches internally; tune batch_size (default 32) for your GPU memory.
• DocumentWriter with policy="overwrite" upserts by Document.id (auto-generated from content hash). For append-only, use policy="duplicate".

Step 8: Build a query pipeline with hybrid retrieval and a prompt builder

Wire the hybrid retriever into a RAG pipeline that feeds a generator. This example uses a local LLM via HuggingFaceLocalGenerator for reproducibility; swap in your provider of choice.

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import HuggingFaceLocalGenerator
from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchHybridRetriever

rag_pipeline = Pipeline()
rag_pipeline.add_component("retriever", hybrid_retriever)
rag_pipeline.add_component("prompt_builder", PromptBuilder(template="""
Answer the question using only the provided context.
Context:
{% for doc in documents %}
  [{{ doc.meta.source_id }}] {{ doc.content }}
{% endfor %}
Question: {{ query }}
Answer:
"""))
rag_pipeline.add_component("llm", HuggingFaceLocalGenerator(model="meta-llama/Llama-3.2-1B-Instruct", task="text-generation"))

rag_pipeline.connect("retriever", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder", "llm")

query = "What retrieval methods does Haystack support with Elasticsearch?"
query_emb = embedder.encode(query).tolist()

result = rag_pipeline.run({
    "retriever": {"query": query, "query_embedding": query_emb, "top_k": 4},
    "prompt_builder": {"query": query}
})
print(result["llm"]["replies"][0])

Verification: The answer should cite doc-2 and doc-3 (the retrieval-category docs) and mention BM25, dense, and hybrid.

Step 9: Observability — logging, metrics, and tracing

You cannot operate a RAG pipeline without visibility into retrieval quality and latency. Add three things:

Structured logging

import logging
import json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if hasattr(record, "extra"):
            log.update(record.extra)
        return json.dumps(log)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.getLogger("haystack").setLevel(logging.INFO)
logging.getLogger("haystack").handlers = [handler]

Retrieval metrics (latency, recall@k)

Wrap retriever calls in a timer and log the top-k scores:

import time
from functools import wraps

def log_retrieval_metrics(retriever):
    @wraps(retriever.run)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = retriever.run(*args, **kwargs)
        elapsed_ms = (time.perf_counter() - start) * 1000
        docs = result.get("documents", [])
        scores = [d.score for d in docs] if docs else []
        logging.info("retrieval_metrics", extra={
            "retriever": retriever.__class__.__name__,
            "latency_ms": round(elapsed_ms, 1),
            "num_docs": len(docs),
            "scores": [round(s, 4) for s in scores],
            "query": kwargs.get("query", "")[:100]
        })
        return result
    return wrapper

hybrid_retriever.run = log_retrieval_metrics(hybrid_retriever)

Distributed tracing (OpenTelemetry)

If you run in Kubernetes or use a managed Elasticsearch, propagate trace context:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317", insecure=True))
)
tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("rag_query") as span:
    span.set_attribute("query", query)
    result = rag_pipeline.run({...})
    span.set_attribute("num_docs_retrieved", len(result["retriever"]["documents"]))

Step 10: Production hardening checklist

Before promoting to prod, verify each item:

Step 11: Common failure modes and fixes

TransportError: 400 - mapper_parsing_exception
Cause: Embedding dimension mismatch. Fix: Recreate index with correct embedding_dim or drop and re-index.

ConnectionError: Connection refused
Cause: Elasticsearch not ready. Fix: Add healthcheck dependency in Docker Compose; implement exponential backoff in client config.

Stale results after write_documents
Cause: Refresh interval. Fix: Call document_store.client.indices.refresh(index="haystack_docs") after bulk writes, or wait for refresh interval.

Low recall on hybrid retriever
Cause: RRF constant too high or low. Fix: Tune rrf_constant (try 30–100); evaluate with a labeled query set.

Metadata filter returns zero hits
Cause: Field mapped as text not keyword. Fix: Verify mapping with GET /haystack_docs/_mapping; ensure filter fields are keyword type.

Step 12: Scaling beyond a single node

When query volume exceeds a single node’s capacity:

1. Add data nodes — Elasticsearch distributes shards automatically. Increase number_of_shards at index creation (cannot change later without reindex).
2. Use search shards — Route queries to a dedicated coordinating node tier.
3. Consider OpenSearch — API-compatible, different licensing, similar operational profile.
4. Offload embeddings — For >10M vectors, evaluate dedicated vector engines (Qdrant, Weaviate, Milvus) with Haystack’s other document stores. Keep Elasticsearch for BM25 + metadata; sync IDs across stores.

You now have a working elasticsearch haystack document store tutorial baseline: provisioned cluster, typed index, hybrid retrieval, metadata filtering, production indexing and query pipelines, and the observability hooks to keep it running. The same patterns apply whether you’re indexing 10K PDFs or 10M — scale the cluster, tune the shards, and keep the pipeline code unchanged.