Elasticsearch With Python: Indexing, Searching, and the Bulk Helper You Should Start With

Q: Why should I create the mapping before indexing?

Dynamic mapping infers a type from the first document it sees, and field types are effectively immutable afterwards. A version string mapped as text cannot be sorted or range-queried, and fixing it requires a full reindex. Defining the mapping up front costs minutes and avoids that.

Q: What is the difference between text and keyword in Elasticsearch?

text is analysed into tokens for matching but cannot be sorted or aggregated. keyword is stored whole and supports filtering, sorting, aggregation, and exact matching but not partial matching. The common pattern is a multi-field declaring both — name for search, name.raw for sorting.

Q: Why does my document not appear in search immediately after indexing?

The index refresh interval defaults to one second, so writes are not instantly searchable. Pass refresh=wait_for on the index call to block until it is visible. Use that in tests only — forcing refresh on every write in production will hurt indexing throughput badly.

Q: What is the right way to bulk index in Python?

Use elasticsearch.helpers.bulk with a generator rather than a list so you can stream data larger than memory, a chunk size around 500 to 1000, and raise_on_error=False so one malformed document does not abandon the rest of the batch. Then inspect the returned error list.

The official elasticsearch-py client connects with Elasticsearch(hosts=[…]), indexes with es.index(), and searches with es.search() — and the one thing to get right before any of that is the mapping.

Elasticsearch will create an index for you on the first write, guessing types as it goes. That convenience is why so many projects end up reindexing three months later: the guess was wrong, mappings are mostly immutable, and by then there are forty million documents built on top of it.

Connecting
Define the mapping before the first write
Indexing documents
Bulk indexing, which is the only sane way to load data
Searching
Paginating past the first ten thousand
Operating it
How this fits the rest of the stack
FAQ

Connecting

pip install elasticsearch

Pin the major version to your cluster. The client refuses to talk to a server a major version away, which is a feature — it prevents a whole class of silent incompatibility — but it means pip install elasticsearch without a constraint will eventually install a client your cluster rejects.

pip install "elasticsearch>=8,<9"

import os
from elasticsearch import Elasticsearch

es = Elasticsearch(
    hosts=[os.environ["ELASTIC_URL"]],
    api_key=os.environ["ELASTIC_API_KEY"],
    request_timeout=30,
    max_retries=3,
    retry_on_timeout=True,
)

print(es.info())

Credentials from the environment, not the source. An API key scoped to the indices this service touches is better than the elastic superuser, for the same reason application database accounts do not get root.

retry_on_timeout is off by default, and turning it on is usually right for reads. Be more careful with writes — a retried index request can produce a duplicate unless you supply an explicit document id.

Define the mapping before the first write

This is the part that pays for itself. Dynamic mapping infers types from the first document it sees, and once a field has a type, you cannot change it. Not with an update, not with a setting. You reindex.

The classic version: a version field arrives as "1.0", gets mapped as text, and now you cannot sort or range-query on it. Or a zip code arrives as 12345, gets mapped as long, and the first document with "01234" fails to index because of the leading zero.

es.indices.create(
    index="products",
    mappings={
        "properties": {
            "sku":         {"type": "keyword"},
            "name":        {"type": "text", "fields": {"raw": {"type": "keyword"}}},
            "price":       {"type": "scaled_float", "scaling_factor": 100},
            "in_stock":    {"type": "boolean"},
            "created_at":  {"type": "date"},
            "description": {"type": "text"},
        }
    },
    settings={"number_of_shards": 1, "number_of_replicas": 1},
)

The text versus keyword distinction is the one to internalise:

text is analysed — broken into tokens, lowercased, stemmed. Good for matching. Cannot be sorted or aggregated on.
keyword is stored whole. Good for filtering, sorting, aggregating, and exact matches. Does not do partial matching.

The name field above declares both: name for searching, name.raw for sorting alphabetically. That multi-field pattern is the standard answer and costs you nothing to set up in advance.

Indexing documents

es.index(
    index="products",
    id="SKU-1029",
    document={
        "sku": "SKU-1029",
        "name": "Wireless keyboard",
        "price": 49.99,
        "in_stock": True,
        "created_at": "2026-08-07T10:30:00Z",
    },
)

Supplying id makes the operation idempotent — indexing the same document twice updates it rather than creating a second copy. Omit it and Elasticsearch generates a random id, so a retry after a network timeout leaves you with duplicates you have no easy way to find.

Use your own primary key as the document id wherever you have one. It makes updates, deletes, and reconciliation against the source of truth straightforward.

Indexing is not immediately visible to search. The refresh interval defaults to one second, which trips up tests that index and immediately query:

es.index(index="products", id="SKU-1", document=doc, refresh="wait_for")

refresh="wait_for" blocks until the document is searchable. Fine in tests. Do not put it in a hot write path — you are trading throughput for immediacy, and forcing refresh=True on every write will flatten indexing performance on a busy index.

Bulk indexing, which is the only sane way to load data

Indexing documents one at a time over HTTP means one round trip per document. For anything past a few hundred, use the bulk helper.

from elasticsearch.helpers import bulk

def generate(rows):
    for row in rows:
        yield {
            "_index": "products",
            "_id": row["sku"],
            "_source": {
                "sku": row["sku"],
                "name": row["name"],
                "price": float(row["price"]),
                "in_stock": row["qty"] > 0,
            },
        }

success, errors = bulk(es, generate(rows), chunk_size=500, raise_on_error=False)
print(f"indexed {success}, failed {len(errors)}")

Two things worth knowing here.

Pass a generator, not a list. The helper consumes lazily, so a generator lets you index a file larger than memory. Materialising ten million dicts into a list first defeats the point.

raise_on_error=False and then check. The default raises on the first failed document and abandons the rest of the batch, which for a nightly load means one malformed row loses the whole run. Collecting errors lets the good documents land and gives you a list to fix.

For very large loads, parallel_bulk runs several threads. Start with bulk and a chunk size around 500 to 1000 documents; that is enough for most workloads, and adding threads before you have measured usually just moves the bottleneck onto the cluster.

Searching

resp = es.search(
    index="products",
    query={
        "bool": {
            "must":   [{"match": {"name": "wireless keyboard"}}],
            "filter": [
                {"term":  {"in_stock": True}},
                {"range": {"price": {"lte": 100}}},
            ],
        }
    },
    sort=[{"price": "asc"}],
    size=20,
)

print(resp["hits"]["total"]["value"])
for hit in resp["hits"]["hits"]:
    print(hit["_score"], hit["_source"]["name"])

The must versus filter split matters for performance. Clauses in must contribute to the relevance score. Clauses in filter do not, which means Elasticsearch can cache them and skip the scoring work entirely.

Anything boolean — in stock, category equals, price under a threshold, date after — belongs in filter. Only the parts where relevance ranking is meaningful belong in must. Putting a term filter in must is not wrong, it is just slower for no benefit.

Note also that total.value caps at 10,000 by default. If you need a true count past that, pass track_total_hits=True and accept the cost.

Paginating past the first ten thousand

from and size work for the first few pages and then stop being viable — deep pagination forces every shard to sort everything up to the offset, and Elasticsearch refuses past 10,000 results by default.

For a user-facing result list, use search_after with a tiebreaker field:

resp = es.search(
    index="products",
    query={"match_all": {}},
    sort=[{"created_at": "asc"}, {"sku": "asc"}],   # tiebreaker keeps it stable
    size=100,
    search_after=last_sort_values,     # from the previous page's final hit
)

For exporting everything, use the scan helper, which handles the cursor for you:

from elasticsearch.helpers import scan

for doc in scan(es, index="products", query={"query": {"match_all": {}}}):
    process(doc["_source"])

Operating it

The Python side is the easy half. Elasticsearch is a stateful, memory-hungry cluster that wants heap tuning, disk headroom, snapshots, and a version upgrade path — and it is unforgiving about running out of disk, which flips indices to read-only and produces confusing write failures.

Whatever runs your indexing and query code, though, is an ordinary service: it needs a runtime that stays up, the cluster URL and API key as environment variables rather than baked into an image, and logs you can read when a bulk load starts failing. Deploying that part on RunxBuild keeps the credentials out of the repository and the deploy history readable, so when search results go strange you can see which build changed the mapping.

One habit worth adopting early: version your index names (products-v3) and point an alias at the current one. When a mapping change forces a reindex — and it will — you build the new index, verify it, and move the alias. No downtime, and a way back if the new mapping is wrong.

How this fits the rest of the stack

Search adds a cluster, and a cluster adds memory, disk, and snapshots to whatever your app was already costing. The RunxBuild hosting calculator puts the runtime, database, storage, and bandwidth line items on one page so you can model the whole shape before committing.

Useful related references:

FAQ

Which Elasticsearch Python client version should I install?

Pin it to your cluster major version, for example pip install elasticsearch>=8,<9. The client deliberately refuses to talk to a server one major version away, so an unpinned install will eventually break when a new major ships.

Why should I create the mapping before indexing?