Exporting 10 Billion Elasticsearch Records: From Simple Script to Enterprise Offline Platform

Problem background

Exporting billions of Elasticsearch documents cannot be handled by a simple three‑step script (query → JSON → upload). At hundred‑million or billion scale the following issues appear:

Deep pagination forces the coordinating node to keep large sorted result sets, increasing CPU and memory pressure.

Loading an entire batch into local memory before encoding leads to OOM.

Single‑process scripts lack orchestration, checkpointing and failure recovery; any network glitch forces a full restart.

Running the export against the online cluster competes with search, recommendation and order‑query workloads for CPU, I/O, heap and segment cache.

Downstream consumers (Hive, Spark, Trino) need partitioned, compressed, replayable files rather than raw JSON.

Real business scenario

An e‑commerce platform needs to export the orders_v3 index for the past 12 months to object storage for three downstream systems: algorithm platform, data warehouse and audit system.

Data volume: ~1 billion documents, raw _source size in the terabyte range.

Export window: overnight, must not noticeably impact the online cluster.

Output format: Parquet + Snappy, partitioned by date.

Semantic: slice‑level retry allowed, but retries must not generate uncontrolled duplicate files.

Recovery: any worker crash must be automatically resumed.

Operations: task progress, throughput, failure reason and hot slices must be observable.

Enterprise‑grade solution architecture

The platform consists of three clear layers:

Control plane : creates tasks, plans slices, schedules slice execution, manages state, configures rate limits and provides operational observability.

Data plane : workers pull data from Elasticsearch, stream‑encode to Parquet, write partitioned files and upload them to object storage.

State plane : persists checkpoints, task status, slice progress and idempotent results.

Key technology stack:

Elasticsearch: PIT + search_after + sliced query Message queue: Kafka (slice task distribution and failure retry)

Metadata store: MySQL (task and slice state)

Cache & checkpoint: Redis (cursor, lease and rate‑control signals)

Output medium: S3/MinIO (partitioned Parquet files)

Runtime: Kubernetes (elastic scaling)

Observability: Prometheus + Grafana + structured logs + TraceId

Common misconceptions

1. Treating export as pagination

Deep pagination requires skipping many documents; query cost grows quickly with page number.

The coordinating node must maintain sorted result sets, inflating CPU and memory usage.

Any failure makes it hard to resume from a stable point.

2. Assuming scroll is the only answer

Scroll keeps a server‑side context; many concurrent exports put sustained pressure on cluster memory.

Scroll lifecycle is tightly coupled to the client connection, making it fragile to network glitches.

When a scroll ID expires, resuming is cumbersome and not truly idempotent.

3. Equating export success with data read

Correct snapshot of data at a point in time.

Each slice must be retryable and recoverable.

Files must be consumable by downstream systems.

Multiple runs must be traceable, auditable and verifiable.

The export must not degrade the online cluster.

Why PIT + search_after + slice?

PIT (Point In Time)

PIT provides a logical, consistent snapshot view for the whole export. It solves two problems:

Without a consistent view, different batches could see different document versions.

If the index keeps receiving writes, batch boundaries would drift.

PIT is lightweight compared with a long‑lived scroll and works naturally with search_after.

search_after for checkpointing

search_after

continues the query from the last sort value of the previous batch. Example sort keys:

create_time asc

_shard_doc asc

or _id asc After each batch the last document’s sort key becomes a natural checkpoint that can be persisted in Redis or MySQL. Advantages:

No large offset maintenance.

Checkpoints are tiny.

Easy to resume.

Naturally compatible with slicing.

Slice for parallel expansion

Parallelism is achieved by splitting a job into independent slices. Benefits:

Each slice is an independent sub‑task with its own cursor and output file.

Workers can scale horizontally by consuming slices.

If a slice fails, only that slice is retried.

From an engineering perspective a slice turns a “big task” into many “small, schedulable, retryable, observable tasks”.

Core data model

Three tables separate job, slice and file manifest information.

CREATE TABLE export_job (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    job_id VARCHAR(64) NOT NULL UNIQUE,
    index_name VARCHAR(128) NOT NULL,
    query_dsl JSON NOT NULL,
    source_fields JSON NOT NULL,
    output_uri VARCHAR(512) NOT NULL,
    output_format VARCHAR(32) NOT NULL,
    snapshot_mode VARCHAR(32) NOT NULL,
    slice_count INT NOT NULL,
    status VARCHAR(32) NOT NULL,
    expected_doc_count BIGINT NULL,
    exported_doc_count BIGINT NOT NULL DEFAULT 0,
    failed_slice_count INT NOT NULL DEFAULT 0,
    created_by VARCHAR(64) NOT NULL,
    created_at DATETIME NOT NULL,
    updated_at DATETIME NOT NULL,
    started_at DATETIME NULL,
    finished_at DATETIME NULL
);

CREATE TABLE export_job_slice (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    job_id VARCHAR(64) NOT NULL,
    slice_id INT NOT NULL,
    lease_owner VARCHAR(128) NULL,
    checkpoint_json JSON NULL,
    status VARCHAR(32) NOT NULL,
    retry_count INT NOT NULL DEFAULT 0,
    exported_doc_count BIGINT NOT NULL DEFAULT 0,
    current_part_no INT NOT NULL DEFAULT 0,
    last_error TEXT NULL,
    started_at DATETIME NULL,
    finished_at DATETIME NULL,
    updated_at DATETIME NOT NULL,
    UNIQUE KEY uk_job_slice(job_id, slice_id)
);

CREATE TABLE export_file_manifest (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    job_id VARCHAR(64) NOT NULL,
    slice_id INT NOT NULL,
    part_no INT NOT NULL,
    object_key VARCHAR(512) NOT NULL,
    row_count BIGINT NOT NULL,
    file_size_bytes BIGINT NOT NULL,
    file_checksum VARCHAR(128) NOT NULL,
    status VARCHAR(32) NOT NULL,
    created_at DATETIME NOT NULL,
    UNIQUE KEY uk_job_slice_part(job_id, slice_id, part_no)
);

Separating these tables enables:

Task‑level overall progress.

Slice‑level hotspot and failure visibility.

File‑level idempotent verification, audit replay and downstream registration.

Full process flow

Normal export flow

Create export task → insert rows into export_job and export_job_slice.

Dispatch slice tasks to Kafka.

Worker consumes a slice, acquires a lease from Redis, opens a PIT.

Loop: fetch next batch with search_after, stream‑encode each document to Parquet, persist checkpoint, update slice progress.

When the writer reaches a roll‑over threshold, flush the current part, upload to object storage, record the part in export_file_manifest.

Renew lease periodically.

After all batches are consumed, close PIT, clear checkpoint, mark slice SUCCESS and refresh job progress.

When all slices succeed, mark job SUCCESS.

Error & recovery flow

Record error and increment retry count.

Release lease; Redis lease expires automatically.

Kafka re‑delivers the slice task (or a scheduler re‑dispatches).

New worker reads the persisted checkpoint and resumes from the exact position.

If retries exceed a threshold, the slice is marked FAILED and the job moves to PARTIAL_FAILED or FAILED for manual handling.

State machine definitions

Task states:

INIT

RUNNING

PARTIAL_FAILED

SUCCESS

FAILED

CANCELLED

Slice states:

PENDING

LEASED

RUNNING

RETRYING

SUCCESS

FAILED

Production‑grade code design (Java / Spring Boot)

Package structure

com.example.export
├── api
│   ├── ExportJobController.java
│   └── dto
├── application
│   ├── ExportJobApplicationService.java
│   ├── SlicePlanner.java
│   └── SliceTaskDispatcher.java
├── domain
│   ├── model
│   ├── repository
│   └── service
├── infrastructure
│   ├── es
│   ├── kafka
│   ├── redis
│   ├── storage
│   └── persistence
└── worker
    ├── ExportSliceConsumer.java
    ├── ExportSliceExecutor.java
    └── ParquetPartWriter.java

Export job creation API

package com.example.export.api;

import com.example.export.api.dto.CreateExportJobRequest;
import com.example.export.api.dto.ExportJobResponse;
import com.example.export.application.ExportJobApplicationService;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/export-jobs")
@RequiredArgsConstructor
public class ExportJobController {
    private final ExportJobApplicationService applicationService;

    @PostMapping
    public ExportJobResponse createJob(@Valid @RequestBody CreateExportJobRequest request,
                                      @RequestHeader("X-Operator") String operator) {
        return applicationService.createJob(request, operator);
    }
}

The request DTO explicitly defines index, query DSL, projected fields, output URI, slice count and batch size:

package com.example.export.api.dto;

import jakarta.validation.constraints.*;
import java.util.List;

public record CreateExportJobRequest(
    @NotBlank String indexName,
    @NotBlank String queryDsl,
    @NotEmpty List<String> sourceFields,
    @NotBlank String outputUri,
    @NotNull @Min(1) Integer sliceCount,
    @NotNull @Min(100) Integer batchSize) {}

Application service (transactional)

package com.example.export.application;

import com.example.export.api.dto.CreateExportJobRequest;
import com.example.export.api.dto.ExportJobResponse;
import com.example.export.domain.model.ExportJob;
import com.example.export.domain.model.ExportJobSlice;
import com.example.export.domain.repository.ExportJobRepository;
import com.example.export.domain.repository.ExportJobSliceRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Service
@RequiredArgsConstructor
public class ExportJobApplicationService {
    private final ExportJobRepository jobRepository;
    private final ExportJobSliceRepository sliceRepository;
    private final SliceTaskDispatcher dispatcher;

    @Transactional
    public ExportJobResponse createJob(CreateExportJobRequest request, String operator) {
        String jobId = "job-" + UUID.randomUUID();
        LocalDateTime now = LocalDateTime.now();
        ExportJob job = ExportJob.newJob(jobId, request.indexName(), request.queryDsl(),
                request.sourceFields(), request.outputUri(), request.sliceCount(),
                request.batchSize(), operator, now);
        jobRepository.save(job);
        List<ExportJobSlice> slices = new ArrayList<>();
        for (int i = 0; i < request.sliceCount(); i++) {
            slices.add(ExportJobSlice.newPending(jobId, i, now));
        }
        sliceRepository.saveAll(slices);
        dispatcher.dispatch(job, slices);
        return ExportJobResponse.from(job);
    }
}

Kafka consumer and slice executor

package com.example.export.worker;

import com.example.export.worker.message.SliceTaskMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class ExportSliceConsumer {
    private final ExportSliceExecutor executor;

    @KafkaListener(topics = "export-slice-task", groupId = "export-worker")
    public void onMessage(SliceTaskMessage message, Acknowledgment ack) {
        try {
            executor.execute(message.jobId(), message.sliceId());
            ack.acknowledge();
        } catch (Exception ex) {
            log.error("slice execute failed, jobId={}, sliceId={}", message.jobId(), message.sliceId(), ex);
            throw ex;
        }
    }
}

Slice executor core logic

package com.example.export.worker;

import com.example.export.domain.model.ExportJob;
import com.example.export.domain.model.ExportJobSlice;
import com.example.export.domain.repository.ExportJobRepository;
import com.example.export.domain.repository.ExportJobSliceRepository;
import com.example.export.infrastructure.es.EsExportClient;
import com.example.export.infrastructure.es.SearchAfterCheckpoint;
import com.example.export.infrastructure.redis.SliceLeaseService;
import com.example.export.infrastructure.redis.SliceCheckpointStore;
import com.example.export.infrastructure.storage.ObjectStorageGateway;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.List;

@Slf4j
@Service
@RequiredArgsConstructor
public class ExportSliceExecutor {
    private final ExportJobRepository jobRepository;
    private final ExportJobSliceRepository sliceRepository;
    private final SliceLeaseService leaseService;
    private final SliceCheckpointStore checkpointStore;
    private final EsExportClient esExportClient;
    private final ObjectStorageGateway objectStorageGateway;
    private final ParquetPartWriterFactory writerFactory;

    public void execute(String jobId, int sliceId) {
        ExportJob job = jobRepository.findByJobId(jobId)
                .orElseThrow(() -> new IllegalArgumentException("job not found"));
        ExportJobSlice slice = sliceRepository.findByJobIdAndSliceId(jobId, sliceId)
                .orElseThrow(() -> new IllegalArgumentException("slice not found"));
        String leaseToken = leaseService.tryAcquire(jobId, sliceId)
                .orElseThrow(() -> new IllegalStateException("slice leased by another worker"));
        try {
            sliceRepository.markRunning(jobId, sliceId);
            SearchAfterCheckpoint checkpoint = checkpointStore.load(jobId, sliceId).orElse(null);
            try (ParquetPartWriter writer = writerFactory.create(job, slice)) {
                String pitId = esExportClient.openPit(job.getIndexName());
                try {
                    while (true) {
                        List<EsDocument> docs = esExportClient.searchNextBatch(job, sliceId, checkpoint, pitId);
                        if (docs.isEmpty()) break;
                        for (EsDocument doc : docs) {
                            writer.write(doc.source());
                        }
                        checkpoint = SearchAfterCheckpoint.fromLast(docs.get(docs.size() - 1));
                        checkpointStore.save(jobId, sliceId, checkpoint);
                        sliceRepository.updateProgress(jobId, sliceId, docs.size(), writer.currentPartNo());
                        if (writer.shouldFlushPart()) {
                            var fileMeta = writer.flushCurrentPart();
                            objectStorageGateway.upload(fileMeta.localPath(), fileMeta.objectKey());
                            sliceRepository.recordPart(jobId, sliceId, fileMeta.partNo(), fileMeta.rowCount());
                        }
                        leaseService.renew(jobId, sliceId, leaseToken);
                    }
                    var finalFile = writer.complete();
                    if (finalFile != null) {
                        objectStorageGateway.upload(finalFile.localPath(), finalFile.objectKey());
                        sliceRepository.recordPart(jobId, sliceId, finalFile.partNo(), finalFile.rowCount());
                    }
                } finally {
                    esExportClient.closePit(pitId);
                }
            }
            checkpointStore.clear(jobId, sliceId);
            sliceRepository.markSuccess(jobId, sliceId);
            jobRepository.refreshJobProgress(jobId);
        } catch (Exception ex) {
            sliceRepository.markRetrying(jobId, sliceId, ex.getMessage());
            throw ex;
        } finally {
            leaseService.release(jobId, sliceId, leaseToken);
        }
    }
}

Checkpoint design

package com.example.export.infrastructure.es;

import java.io.Serializable;
import java.util.List;

public record SearchAfterCheckpoint(List<Object> sortValues) implements Serializable {
    public static SearchAfterCheckpoint fromLast(EsDocument document) {
        return new SearchAfterCheckpoint(document.sortValues());
    }
}

Parquet writer considerations

Key points:

File rolling by row count or byte size.

Schema evolution handling (missing fields, nulls, type compatibility).

Temporary file naming to avoid overwriting during retries.

A part is considered finished only after successful upload and manifest registration.

Suggested object key pattern:

s3://offline-lake/es/orders_v3/dt=2026-08-06/job_id=job-xxx/slice=0003/part-0007.parquet

Concurrency & scalability design

Export throughput is limited by four resource categories: Elasticsearch query capacity, worker CPU/heap, local disk write, and object‑storage upload bandwidth. Concurrency control is layered:

Task‑level: limit how many export jobs run concurrently.

Slice‑level: limit concurrent slices per job.

Worker‑level: fixed thread pool for slice executors.

Request‑level: token‑bucket rate limiting for Elasticsearch queries.

Recommended practices:

Kafka partitions ≥ maximum slice count.

Each worker uses a fixed thread pool for slice execution.

Within a slice processing is strictly serial; no intra‑slice parallelism.

Separate thread pools for ES fetch and object‑storage upload to avoid blocking.

Resource isolation pools: slice-fetch-pool: handles Elasticsearch pulls. file-upload-pool: handles object‑storage uploads. job-control-pool: aggregates task status and performs compensation.

In Kubernetes, auto‑scaling should consider both Kafka consumer lag ( kafka_consumer_lag) and the number of running slices ( export_running_slices).

Reliability design

Idempotency

Slice lease guarantees exclusive execution.

Each part file is uniquely identified by ( jobId, sliceId, partNo).

Only after successful upload is the manifest entry created; downstream reads only READY files.

Retry strategy

Retriable errors (network timeout, temporary ES rejection, transient storage failure, worker eviction) → exponential back‑off, up to N attempts.

Non‑retriable errors (invalid DSL, schema incompatibility, permission missing, unwritable target) → slice marked FAILED, task escalates to manual handling.

After repeated failures, write the latest checkpoint and error snapshot to a dead‑letter queue.

Failure takeover

Redis lease expires automatically.

Kafka or a scheduler re‑dispatches the slice.

New worker reads the persisted checkpoint and continues from the exact position.

Performance optimisation

Elasticsearch side

Prefer a read‑only replica cluster or an offline query cluster to avoid impacting production workloads.

Project only required fields; avoid full _source transfer.

Batch size tuning: too small → RPC overhead; too large → memory and encoding cost.

Slice count balancing: too many slices increase coordination overhead, Kafka message volume and worker context switches.

Worker side

Stream‑write each batch directly to Parquet; avoid aggregating large in‑memory collections.

Reuse schema conversion objects, minimise large Map allocations and string concatenations to reduce GC pressure.

File rolling by row count or byte size prevents oversized files that slow upload and downstream reads.

Object‑storage side

Use multipart upload with appropriate part size.

Dedicated upload thread pool separate from fetch pool.

Plan bucket prefixes to avoid hotspot concentration.

Security & governance

Permission checks for who can create export jobs, access specific indices and write to particular object‑storage paths.

Sensitive fields (phone, ID, email, address) support three policies: block export, mask during export, or encrypt before export.

All critical configuration (ES endpoints, auth, rate limits, default slice count, batch size, Parquet thresholds, bucket prefixes) must be externalised per environment (dev / test / prod).

Observability

Metrics (Prometheus) to instrument:

export_job_running_total

export_slice_running_total

export_slice_retry_total

export_docs_exported_total

export_part_upload_seconds

export_es_query_seconds

export_checkpoint_save_fail_total

export_object_upload_fail_total

Structured logs must contain at least: jobId, sliceId, traceId, pitId (or its hash), partNo, checkpoint, errorCode Typical alerts:

Slice progress stalls.

Task retry count spikes.

Elasticsearch query P99 latency surge.

Object‑storage upload failure rate increase.

Kafka lag accumulation.

Kubernetes deployment example

apiVersion: apps/v1
kind: Deployment
metadata:
  name: export-worker
spec:
  replicas: 6
  selector:
    matchLabels:
      app: export-worker
  template:
    metadata:
      labels:
        app: export-worker
    spec:
      containers:
        - name: export-worker
          image: registry.example.com/export-worker:1.0.0
          env:
            - name: SPRING_PROFILES_ACTIVE
              value: prod
          resources:
            requests:
              cpu: "1"
              memory: "2Gi"
            limits:
              cpu: "2"
              memory: "4Gi"
          volumeMounts:
            - name: tmp-dir
              mountPath: /data/tmp
      volumes:
        - name: tmp-dir
          emptyDir:
            sizeLimit: 20Gi

Important notes:

Local temporary volume must have a size limit to prevent a runaway task from filling the node disk.

Upload thread pool size should be bounded; otherwise it can consume all CPU.

For auto‑scaling based on Kafka lag, integrate KEDA or custom HPA metrics such as kafka_consumer_lag and export_running_slices.

Error scenarios & applicability boundaries

Typical wrong approaches and why they fail:

Exporting to a single massive JSON file → OOM, unrecoverable failures, poor downstream performance.

Single‑node multithreaded scan of the production cluster → lacks global flow control, no platform‑level observability, hard to hand over on failure.

Keeping export state only in memory → a process crash loses all progress; no cross‑instance hand‑over.

Slice without idempotency → retries generate duplicate files, downstream cannot determine the final result set.

Suitable for:

Billion‑scale or multi‑TB Elasticsearch offline export where checkpointing and platform governance are required.

Downstream systems that consume data lakes, warehouses or batch processors.

Not ideal for:

Few‑million one‑off exports, low‑risk manual jobs, or scenarios where real‑time CDC, log subscription or stream processing would be a better fit.

Future evolution paths

Full + incremental export integration to avoid re‑scanning already exported data.

Hot‑cold tiered export: high‑frequency incremental export for hot data, low‑frequency large‑batch archival for cold data.

Unified extraction platform covering Elasticsearch, MySQL, ClickHouse, MongoDB, etc., instead of per‑source scripts.

Orchestration layer adding validation, merge, partition registration, metadata enrollment and downstream notifications, turning the platform into a full offline data orchestration foundation.

Final takeaway

Exporting at the billion‑record scale tests consistency, task decomposition, impact on the online cluster, idempotent retry handling and observability. A mature solution combines PIT + search_after + slicing, a three‑layer architecture, Kafka‑Redis‑MySQL glue, and robust operational practices (metrics, structured logs, alerts, Kubernetes auto‑scaling). The journey from a simple script to an enterprise‑grade platform is essentially a journey toward building reliable offline data capabilities.