Designing Batch Reindex Workflows #

stateDiagram-v2
  [*] --> Provision
  Provision --> SubmitReindex
  SubmitReindex --> Polling
  Polling --> Polling: in progress
  Polling --> Validate: completed
  Validate --> AliasSwap: counts match
  Validate --> Investigate: failures present
  AliasSwap --> [*]
  Investigate --> SubmitReindex: bounded retry

Concept: Deterministic State Machines Over Ad-Hoc Copies #

Production Elasticsearch environments rarely tolerate monolithic _reindex executions. Data migration during mapping evolution, shard rebalancing, or storage tiering must operate as an atomic, observable pipeline rather than a blind bulk copy. Designing Batch Reindex Workflows requires shifting from manual console invocations to deterministic state machines that enforce idempotent execution, bounded concurrency, and explicit failure boundaries. By chunking large indices into deterministic scroll windows, applying document-level transformations, and committing results in controlled bulk batches, engineers prevent thread pool exhaustion, mitigate circuit breaker trips, and guarantee that partial failures never corrupt target index state. This methodology directly extends the operational patterns established in Automated Reindexing Pipelines & Workflows by treating reindexing as a controlled ILM transition rather than a one-off data movement.

Pre-flight Configuration: Templates, Allocation, and Thresholds #

Reindex success is dictated before the first document is copied. The target index must be provisioned with explicit mapping definitions, optimized shard counts, and attached ILM policies that mirror or supersede the source lifecycle. Shard allocation awareness is non-negotiable: routing rules and index.routing.allocation.require tags must be configured to prevent hot-warm-cold tier violations during ingestion.

PUT _index_template/reindex_target_template
{
  "index_patterns": ["logs-v2-*"],
  "template": {
    "settings": {
      "number_of_shards": 5,
      "number_of_replicas": 1,
      "index.routing.allocation.include._tier_preference": "data_hot",
      "index.lifecycle.name": "logs-retention-90d",
      "index.lifecycle.rollover_alias": "logs-v2"
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "message": { "type": "text", "ignore_malformed": true, "coerce": true }
      }
    }
  }
}

ILM policy attachment must occur immediately after index creation but before data ingestion begins. This ensures rollover, shrink, and delete phases are governed from the first document. Mapping updates should explicitly evaluate ignore_malformed and coerce against historical data quality to prevent ingestion halts. When calibrating bulk execution parameters, baseline against cluster write throughput under peak query load. Target 500–2000 documents per batch and set requests_per_second to 30–50% of sustained write capacity. Overly aggressive scroll windows trigger circuit_breaking_exception errors, while undersized batches increase network overhead and prolong alias transition windows.

Python v8+ Orchestration: Async Execution & Idempotent Pipelines #

Production-grade reindexing demands programmatic control over the Elasticsearch REST API. The elasticsearch-py v8+ client provides async execution, automatic retry handling, and structured response parsing that replaces brittle shell scripts. The following orchestration pattern initializes an async client, submits a throttled _reindex task, and polls for completion with exponential backoff.

import asyncio
import logging
from elasticsearch import AsyncElasticsearch
from elasticsearch.exceptions import ApiError

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")

async def execute_batch_reindex(es: AsyncElasticsearch, source: str, target: str, rps: int = 1000):
    """
    Submits a throttled, idempotent reindex task and returns the task_id.
    Uses op_type=create to enforce idempotency and skip existing documents.
    """
    reindex_body = {
        "source": {
            "index": source,
            "size": 1000,
            "query": {"match_all": {}}
        },
        "dest": {
            "index": target,
            "op_type": "create"
        },
        "max_docs": 5000000,
        "requests_per_second": rps,
        "conflicts": "proceed"
    }

    try:
        response = await es.reindex(body=reindex_body, wait_for_completion=False)
        task_id = response["task"]
        logging.info(f"Reindex task submitted: {task_id}")
        return task_id
    except ApiError as e:
        logging.error(f"Failed to submit reindex task: {e.info}")
        raise

async def poll_task_progress(es: AsyncElasticsearch, task_id: str, interval: int = 5):
    """
    Polls the _tasks API until completion or failure.
    """
    while True:
        try:
            task = await es.tasks.get(task_id=task_id)
            status = task["task"]["status"]
            completed = task["completed"]
            
            logging.info(
                f"Progress: {status['created']}/{status['total']} docs | "
                f"Updated: {status['updated']} | Conflicts: {status['version_conflicts']}"
            )
            
            if completed:
                # A task-level failure surfaces as a top-level `error`; per-document
                # failures appear under response.failures (only on a completed task).
                if "error" in task:
                    logging.error(f"Task failed: {task['error']}")
                    raise RuntimeError("Reindex task failed")
                failures = task.get("response", {}).get("failures", [])
                if failures:
                    logging.error(f"Reindex finished with {len(failures)} document failure(s).")
                    raise RuntimeError("Reindex task had document-level failures")
                logging.info("Reindex task completed successfully.")
                return status
            
            await asyncio.sleep(interval)
        except ApiError as e:
            logging.warning(f"Task polling interrupted: {e}. Retrying in {interval*2}s")
            await asyncio.sleep(interval * 2)
            interval = min(interval * 2, 30)

Idempotency is enforced via op_type: "create", which skips documents that already exist in the target index. When handling version collisions or duplicate primary keys during migration, consult Resolving Document Conflicts During Reindex for conflict routing strategies. The full implementation for zero-downtime alias swapping and pipeline handoff is detailed in Python Script for Zero-Downtime Elasticsearch Reindexing.

Threshold Tuning & Circuit Breaker Mitigation #

Reindex throughput is constrained by three cluster boundaries: JVM heap, disk I/O, and the bulk thread pool queue. The requests_per_second parameter acts as a token-bucket throttle, but it does not prevent circuit breaker trips if individual scroll payloads exceed indices.breaker.total.limit.

To safely scale batch operations:

1. Scroll Size: Keep source.size between 1000–2000. Larger payloads increase deserialization overhead and trigger circuit_breaking_exception on coordinating nodes.
2. Thread Pool Queue: Monitor thread_pool.write.queue. If queue depth consistently exceeds 50% of thread_pool.write.size, reduce requests_per_second or increase index.refresh_interval to -1 during ingestion.
3. Heap Pressure: Set indices.breaker.fielddata.limit and indices.breaker.request.limit to conservative values (typically 40–50% of heap) during migration. Use _nodes/stats/breaker to validate thresholds before execution.

For continuous calibration methodologies, reference Optimizing Reindex Thresholds & Bulk Sizes to align batch parameters with cluster topology.

Real-World Debugging & Failure Boundaries #

Even with throttling and idempotency guards, production reindex operations encounter edge cases. The following debugging flows map directly to observable cluster metrics:

Progress tracking should never rely on console output alone. Query _tasks?detailed=true&actions=*reindex to extract created, updated, deleted, and batches counters. For automated monitoring integration, see Tracking Reindex Progress & Performance.

When debugging async Python clients, leverage the elasticsearch client’s built-in retry logic via retry_on_timeout=True and max_retries=3. Always wrap bulk operations in try/except ApiError blocks and log e.info["error"]["caused_by"] for precise failure attribution. The official Elasticsearch Python Client Documentation provides comprehensive async configuration patterns, while Python’s asyncio library governs event loop management for high-concurrency polling.

Final validation requires a document count audit and a sampling checksum. Execute GET /_cat/indices/source,target?v to verify parity, then run a scripted _search with track_total_hits: true and routing constraints to confirm shard-level consistency before alias promotion.