Elasticsearch — Python Client and Production Pitfalls

elasticsearch-py (or opensearch-py for the Amazon fork). Production patterns: bulk ops, retries, async, and the gotchas that bite real systems.

Basic client

from elasticsearch import Elasticsearch

es = Elasticsearch(
    "https://my-cluster.eu-west-1.aws.found.io:9243",
    basic_auth=("elastic", "password"),
    verify_certs=True,
    request_timeout=30,
    retry_on_timeout=True,
    max_retries=3,
)

# Indexing
es.index(index="users", id="u1", document={"name": "Alice", "age": 30})

# Search
result = es.search(index="users", query={"match": {"name": "Alice"}})

# Get by ID
doc = es.get(index="users", id="u1")

Single client, long-lived. Connection-pool managed internally.

Async client

from elasticsearch import AsyncElasticsearch

es = AsyncElasticsearch("https://...")

async def search():
    return await es.search(index="users", query={"match_all": {}})

Uses aiohttp under the hood. Same API as sync.

Bulk indexing — mandatory for performance

Single index() calls have ~5-20ms latency. Bulk in batches:

from elasticsearch.helpers import bulk

actions = [
    {"_index": "users", "_id": user.id, "_source": user.dict()}
    for user in users
]

success, errors = bulk(es, actions, chunk_size=500, request_timeout=60)

Wins:

• One HTTP request per chunk vs one per doc.
• Lower per-doc CPU cost on the server (parsing).
• 10-100× faster ingestion.

Async version: async_bulk from elasticsearch.helpers.

Bulk with errors

from elasticsearch.helpers import bulk, BulkIndexError

try:
    bulk(es, actions, raise_on_error=True)
except BulkIndexError as e:
    for err in e.errors:
        # err is the per-doc failure dict
        log.error("Indexing failed", item=err)

With raise_on_error=False, the function returns (success_count, list_of_errors) — you can keep going and log failures.

Pagination patterns

Three modes:

search_after

result = es.search(index="users", size=20, sort=[{"created_at": "desc"}, {"_id": "asc"}])
last = result["hits"]["hits"][-1]["sort"]

# Next page
next_page = es.search(
    index="users", size=20,
    sort=[{"created_at": "desc"}, {"_id": "asc"}],
    search_after=last
)

Stateless cursor. Constant cost per page.

Point-in-Time + search_after

For long-running scrolls (export 10M docs):

pit = es.open_point_in_time(index="users", keep_alive="5m")
pit_id = pit["id"]

# Loop
while True:
    result = es.search(
        size=1000,
        sort=[{"_id": "asc"}],
        pit={"id": pit_id, "keep_alive": "5m"},
        search_after=last_sort,
    )
    if not result["hits"]["hits"]:
        break
    last_sort = result["hits"]["hits"][-1]["sort"]
    process(result["hits"]["hits"])

es.close_point_in_time(id=pit_id)

PIT snapshots the index state. Concurrent writes don’t disturb your scroll.

Replaces the older scroll API (deprecated in favor of PIT + search_after).

Aliases and reindex pattern

Production: never read/write the underlying index directly. Use aliases:

es.indices.create(index="users-v1", body={"mappings": {...}})
es.indices.put_alias(index="users-v1", name="users")

# App reads/writes via alias "users"
es.index(index="users", id="u1", document={...})

# To reindex with a new mapping:
es.indices.create(index="users-v2", body={"mappings": {...}})
es.reindex(body={"source": {"index": "users-v1"}, "dest": {"index": "users-v2"}}, wait_for_completion=False)
# Poll the task to monitor progress

# Atomic alias swap when reindex completes
es.indices.update_aliases(body={"actions": [
    {"remove": {"index": "users-v1", "alias": "users"}},
    {"add":    {"index": "users-v2", "alias": "users"}},
]})

Aliases enable:

• Zero-downtime reindex (atomic swap).
• Multi-index search (one alias → multiple time-series indices).
• Index rotation (write to current; read from current + archived).

Mapping evolution — only adding fields works

To change a field type, you reindex. No update-mapping-in-place for type changes.

Workflow:

1. Create new index with corrected mapping.
2. Reindex.
3. Swap aliases.
4. Optionally delete old index.

Refresh interval — tune for bulk

# Before bulk load
es.indices.put_settings(index="users", body={"index": {"refresh_interval": "-1", "number_of_replicas": 0}})

# Bulk load
bulk(es, actions, chunk_size=1000)

# After
es.indices.put_settings(index="users", body={"index": {"refresh_interval": "1s", "number_of_replicas": 1}})
es.indices.forcemerge(index="users", max_num_segments=1)

refresh_interval: -1 disables periodic refresh during load (much faster). Replicas at 0 means no replication overhead. Re-enable after.

Index Lifecycle Management (ILM)

For time-series data: rotate indices automatically by age / size.

PUT /_ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot":    { "actions": { "rollover": { "max_size": "50gb", "max_age": "30d" } } },
      "warm":   { "min_age": "30d", "actions": { "shrink": { "number_of_shards": 1 }, "forcemerge": { "max_num_segments": 1 } } },
      "cold":   { "min_age": "60d", "actions": { "searchable_snapshot": {} } },
      "delete": { "min_age": "365d", "actions": { "delete": {} } }
    }
  }
}

Hot (active writes) → warm (read-only, smaller shards) → cold (archived to S3) → delete. Standard for log / metrics ingestion at scale.

Multi-tenant indexing

Three patterns:

1. One index per tenant — strong isolation. Caps at ~thousands of tenants (cluster-state size).
2. One index, tenant filter — single index, term: {tenant_id: ...} filter on every query. Scales to many tenants.
3. Hybrid — index-per-large-tenant, shared index for small tenants.

For most cases: shared index with tenant filter. Use a routing key for distribution:

es.index(index="users", id="u1", document={"tenant": "acme", "name": "Alice"}, routing="acme")
es.search(index="users", routing="acme", query={"match": {"name": "Alice"}})

Routing localizes the tenant’s data to one shard — faster queries, but watch for hot shards if one tenant dominates.

Common production gotchas

Mapping explosion

Dynamic mapping + user-provided keys + millions of unique fields → cluster state blows up, node startup takes forever. Use flattened type or strict mappings.

Shard size

Rule of thumb: 10-50 GB per shard. Too small (1 KB shards) wastes overhead; too large (200 GB) hurts query latency, rebalance time, and recovery.

Calculate: total data / 30 GB ≈ shard count.

Too many shards

A 100-node cluster with 10,000 shards has 100k shard pings flying around. Cluster state contention; slow operations.

Rule: shards per node ≤ 20 × number of CPUs. For a typical box, 20-100 shards.

Hot shards

One tenant or partition keeps getting all writes → one shard saturates while others idle. Mitigations:

• Route consciously.
• Pre-split heavy tenants.
• Custom routing strategy.

Deep paginate without realizing

from: 9000, size: 100 triggers each shard returning 9100 docs to the coordinator. Slow.

ES enforces max_result_window: 10000 by default. Beyond: use search_after.

Aggregation OOM

Terms agg on a high-cardinality field with size: 1000000 → tries to load 1M buckets into memory per shard. Use composite for paginated grouping.

Wrong field type

User indexed {"price": "9.99"} (string) before the explicit mapping was applied. Future int-typed docs fail. Reindex.

Single-cluster-as-source-of-truth

Elasticsearch is not your primary DB. Treat it as a read replica of your real data store. If it gets corrupted, reindex from source.

Monitoring essentials

Common patterns checklist

• Always read/write via aliases.
• Bulk ingest with helpers.bulk (or async_bulk).
• Disable refresh + replicas for big initial loads; re-enable after.
• Explicit mappings (never rely on dynamic for production).
• keyword for IDs / enums; text for searchable text; multi-fields when both needed.
• search_after (not from/size) for deep pagination.
• routing for multi-tenant isolation on shared indices.
• ILM for time-series data lifecycle.

Interview angle

• “How do you index a million documents fastest?” — helpers.bulk in chunks of ~500-1000, with refresh_interval: -1, replicas at 0, single-threaded per partition. Re-enable refresh + replicas after.
• “Pagination beyond 10,000 results?” — search_after for stateless cursors; _pit + search_after for snapshot iteration (long-running exports). from/size doesn’t scale past max_result_window.
• “How do you change a field type?” — you don’t change it in place. Create a new index with the corrected mapping, reindex (via _reindex API, often with wait_for_completion: false + task polling), then atomic alias swap.
• “Reindex strategy for a 100 GB index?” — _reindex with slices for parallelism, wait_for_completion: false, poll the task API. Or do it externally (read from old, bulk to new) for more control. Alias-swap at the end.
• “How do you do multi-tenant?” — shared index with tenant_id filter is the scalable default. Optional routing parameter to localize a tenant’s docs to one shard (faster but watch for hot shards). One-index-per-tenant only for true isolation needs and few tenants.
• “Cluster status is yellow — what does it mean?” — primary shards are assigned but some replicas aren’t. Cluster works for reads/writes but redundancy is compromised. Common cause: not enough nodes for replication factor; or a node temporarily down.
• “How do you avoid mapping explosions?” — explicit mappings (dynamic: strict), flattened type for arbitrary user keys, index.mapping.total_fields.limit as a safety net. Never dynamically index user-supplied object keys at scale.