Elasticsearch: Query DSL, Aggregations, kNN and Hybrid Search Reference Guide

# Docker — single-node dev cluster with security off (do NOT do this in prod)
docker run -d --name es \
    -p 9200:9200 \
    -e "discovery.type=single-node" \
    -e "xpack.security.enabled=false" \
    docker.elastic.co/elasticsearch/elasticsearch:8.14.0

# Smoke test — should return cluster info
curl -s http://localhost:9200 | jq

# Cluster health
curl -s 'http://localhost:9200/_cluster/health?pretty'

# Kibana (optional dev UI)
docker run -d --name kib --link es \
    -p 5601:5601 \
    -e ELASTICSEARCH_HOSTS=http://es:9200 \
    docker.elastic.co/kibana/kibana:8.14.0

# Python client (matches the server major version)
pip install "elasticsearch>=8.14,<9"

python - <<'PY'
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200")
print(es.info()["version"]["number"])
PY

# Explicit mapping for a blog index — custom analyzer + multi-fields + a dense_vector.
PUT blogs
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "analysis": {
      "filter": {
        "english_stop":  { "type": "stop",    "stopwords": "_english_" },
        "english_stem":  { "type": "stemmer", "language": "english"    }
      },
      "analyzer": {
        "blog_text": {
          "tokenizer": "standard",
          "filter": ["lowercase", "english_stop", "english_stem"]
        }
      },
      "normalizer": {
        "lc_norm": { "type": "custom", "filter": ["lowercase", "asciifolding"] }
      }
    }
  },
  "mappings": {
    "dynamic": "strict",
    "properties": {
      "title":   { "type": "text", "analyzer": "blog_text",
                   "fields": { "raw": { "type": "keyword", "normalizer": "lc_norm" } } },
      "body":    { "type": "text", "analyzer": "blog_text", "index_options": "offsets" },
      "tags":    { "type": "keyword" },
      "author":  { "type": "keyword" },
      "ts":      { "type": "date", "format": "strict_date_optional_time||epoch_millis" },
      "views":   { "type": "long" },
      "embedding": { "type": "dense_vector", "dims": 768, "similarity": "cosine" }
    }
  }
}

# Hybrid bool query — full-text "must" scored by BM25,
# filters (don't affect score), recency decay, paginated + sorted.
POST blogs/_search
{
  "size": 20,
  "from": 0,
  "track_total_hits": false,
  "_source": ["title", "author", "ts"],
  "query": {
    "function_score": {
      "query": {
        "bool": {
          "must":   [ { "multi_match": { "query": "vector search redis",
                                         "fields": ["title^3", "body"], "type": "best_fields" } } ],
          "filter": [ { "term":  { "author": "abhishek" } },
                      { "range": { "ts": { "gte": "now-30d/d" } } } ],
          "must_not": [ { "term": { "tags": "draft" } } ]
        }
      },
      "functions": [
        { "gauss": { "ts": { "origin": "now", "scale": "7d", "decay": 0.5 } } }
      ],
      "score_mode": "multiply",
      "boost_mode": "multiply"
    }
  },
  "sort": [ "_score", { "ts": "desc" } ],
  "highlight": { "fields": { "title": {}, "body": {} } }
}

# Hybrid search: BM25 + dense_vector kNN combined via Reciprocal Rank Fusion (RRF).
# Each retriever returns its own top-k; RRF blends the rankings — no score tuning required.
POST blogs/_search
{
  "size": 10,
  "_source": ["title", "tags", "ts"],
  "retriever": {
    "rrf": {
      "rank_window_size": 50,
      "rank_constant": 60,
      "retrievers": [
        { "standard": {
            "query": {
              "bool": {
                "must":   [ { "multi_match": { "query": "vector retrieval",
                                               "fields": ["title^3", "body"] } } ],
                "filter": [ { "term":  { "tags": "search" } },
                            { "range": { "ts": { "gte": "now-180d/d" } } } ]
              }
            }
        }},
        { "knn": {
            "field": "embedding",
            "query_vector": [/* 768 floats from your embedding model */],
            "k": 30,
            "num_candidates": 200,
            "filter": [ { "term": { "tags": "search" } } ]
        }}
      ]
    }
  }
}

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk, scan

es = Elasticsearch("https://es.internal:9200", api_key="****", verify_certs=True)

# Stream actions; helpers.bulk batches them server-side.
def gen_actions(rows):
    for r in rows:
        yield {
            "_op_type": "index",          # index | create | update | delete
            "_index":   "blogs",
            "_id":      r["id"],          # deterministic ids => idempotent
            "_source":  {
                "title":     r["title"],
                "body":      r["body"],
                "tags":      r["tags"],
                "ts":        r["ts"],
                "embedding": r["embedding"],
            },
        }

ok, errors = bulk(
    es, gen_actions(rows),
    chunk_size=500,                       # per-batch
    request_timeout=60,
    raise_on_error=False,                 # let caller decide
    refresh=False,                        # async refresh = throughput
)
print(f"{ok} indexed, {len(errors)} errors")

# Stream every doc back out — uses scroll API under the hood.
for hit in scan(es, index="blogs", query={"query": {"match_all": {}}}, size=1000):
    handle(hit["_source"])

# End-to-end: create a blog index with text + vector fields,
# bulk-index a few rows, run a hybrid (BM25 + kNN) search with a filter.
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

es = Elasticsearch("http://localhost:9200")

# 1. Mapping with text + dense_vector
es.indices.delete(index="blogs", ignore_unavailable=True)
es.indices.create(index="blogs", mappings={
    "properties": {
        "title":     {"type": "text"},
        "body":      {"type": "text"},
        "tags":      {"type": "keyword"},
        "embedding": {"type": "dense_vector", "dims": 4, "similarity": "cosine"},
    }
})

# 2. Bulk index
docs = [
    {"id": 1, "title": "Vector search", "body": "kNN in Elasticsearch.",
     "tags": ["search"], "embedding": [0.1, 0.2, 0.3, 0.4]},
    {"id": 2, "title": "Redis caching", "body": "Hot keys, TTLs.",
     "tags": ["cache"],  "embedding": [0.0, 0.1, 0.0, 0.9]},
]
bulk(es, ({"_index": "blogs", "_id": d["id"], "_source": d} for d in docs), refresh="wait_for")

# 3. Hybrid query — BM25 must + kNN, filter by tag
hits = es.search(index="blogs", size=5, query={
    "bool": {
        "must":   [{"match": {"body": "knn"}}],
        "filter": [{"term":  {"tags": "search"}}]
    }
}, knn={
    "field":          "embedding",
    "query_vector":   [0.1, 0.2, 0.25, 0.5],
    "k":              5, "num_candidates": 20
})
for h in hits["hits"]["hits"]:
    print(h["_score"], h["_source"]["title"])