Ingest Pipelines: Data Transformation Before Indexing

1. Where data transformation should happen: application or index

Every data source delivers raw data in a format that rarely matches the target schema of the search index exactly. Prices arrive as strings with thousands separators, category assignments as comma-separated free-text fields, descriptions as unstructured running text with embedded technical specifications. The classic answer is to handle this cleanup in application code before a document is even sent to Elasticsearch, for example in a Magento indexer or a separate export script that reads product data from the database and maps it onto the target schema.

That application-side solution is not wrong, but it moves the logic entirely into custom code that has to be maintained, tested, and adjusted with every schema change. Ingest pipelines offer an alternative directly inside Elasticsearch: the transformation logic lives declaratively as a pipeline definition in the cluster, can be changed independently of the sending application, and automatically applies to every document that passes through that pipeline, regardless of which source it came from.

2. Ingest pipelines: an ordered chain of processors

An ingest pipeline is a named sequence of processors, defined through the ingest API, that a document passes through one after another before it is actually written to a shard. Each processor receives the document in its current intermediate state, changes fields, adds new ones, or removes existing ones, and passes the result on to the next processor. This processing runs on the node coordinating the write, before the actual indexing logic takes over.

A pipeline is referenced either explicitly via the pipeline parameter on an index or bulk request, or implicitly through the index setting index.default_pipeline, which automatically applies to every document in that index. There is also index.final_pipeline, which always runs last, regardless of which pipeline was explicitly specified beforehand, for example to always set a timestamp or a version marker regardless of the source.

PUT _ingest/pipeline/product-basic-cleanup
{
  "description": "Trim, lowercase and default currency for product docs",
  "processors": [
    { "trim": { "field": "sku" } },
    { "lowercase": { "field": "sku" } },
    { "set": { "field": "currency", "value": "EUR", "override": false } }
  ]
}

3. The grok processor: parsing unstructured strings into named fields

The grok processor parses unstructured text into named fields using predefined and custom patterns, a concept many will recognize from Logstash. A pattern like %{IP:client_ip} %{WORD:method} %{URIPATHPARAM:request} splits a log line into individual, searchable, and aggregatable fields instead of storing it as one opaque string. Elasticsearch ships an extensive library of ready-made patterns for common formats such as Apache logs, syslog, or IP addresses, and custom patterns can be added.

In a product context, grok fits semi-structured manufacturer fields well, for example a technical designation like Art.-Nr. 4711-XL-BLAU, from which the SKU, size, and color can be extracted as separate, filterable fields. If the pattern fails to match a document because the format deviates from expectations, grok tags the document by default in the _ingest._value.tags field instead of failing the entire write.

{
  "grok": {
    "field": "raw_article_code",
    "patterns": ["Art\\.-Nr\\. %{DATA:article_number}-%{DATA:size}-%{WORD:color}"],
    "tag": "parse-article-code"
  }
}

4. The script processor: Painless for custom transformation logic

Where prebuilt processors such as convert, split, or rename fall short, the script processor runs arbitrary Painless logic directly inside the pipeline. This suits cases that cannot be expressed with a single declarative instruction, for example a conditional price calculation depending on several fields at once, or normalization that applies different rules depending on the source system.

It matters to use the script processor deliberately and sparingly, since every line of code runs again for every document and therefore directly affects write throughput. For simple renames, type conversions, or setting default values, dedicated, considerably cheaper processors exist that accomplish the same task without the overhead of Painless evaluation.

{
  "script": {
    "source": "ctx.price_normalized = Double.parseDouble(ctx.price_raw.replace(',', '.'))"
  }
}

5. The enrich processor: pulling reference data from another index

The enrich processor performs a lookup against a prepared reference dataset while a document is being written and adds matching fields, comparable to a left join at indexing time. It requires an enrich policy that defines which source index serves as the reference, which match field drives the lookup, and which target fields get copied over. After the policy runs, Elasticsearch builds an optimized, hidden enrich index from it.

A typical example is a manufacturer reference: a product document only contains a brand ID, and the enrich processor loads the full brand name, country of origin, and a logo shorthand at indexing time and writes those fields directly into the product document. That way the searching application no longer needs a separate call to load this extra data, it already sits denormalized inside the searchable document.

PUT _enrich/policy/brand-lookup
{
  "match": {
    "indices": "brand-reference",
    "match_field": "brand_id",
    "enrich_fields": ["brand_name", "brand_country", "brand_logo_code"]
  }
}
POST _enrich/policy/brand-lookup/_execute

{
  "enrich": {
    "policy_name": "brand-lookup",
    "field": "brand_id",
    "target_field": "brand"
  }
}

6. Practical example: a complete pipeline for product data normalization

In practice, a single product pipeline usually combines several processors: trim and lowercase normalize the SKU, convert turns a price delivered as a string into a numeric type, split breaks a comma-separated category list into an array, and the enrich processor adds brand data from the reference table. Each processor solves exactly one clearly scoped problem instead of a single, hard-to-maintain script handling everything.

Breaking things down into individual, named processors also makes the pipeline understandable to colleagues who did not write it themselves, and lets individual steps be tested or swapped out without rewriting the entire transformation logic. For a Magento product export, that means the export logic in the application stays as close to the database's raw format as possible, while the pipeline handles the actual normalization onto the search schema.

PUT _ingest/pipeline/product-normalize
{
  "processors": [
    { "trim": { "field": "sku" } },
    { "lowercase": { "field": "sku" } },
    { "convert": { "field": "price_raw", "type": "float", "target_field": "price" } },
    { "split": { "field": "categories_raw", "separator": "," } },
    { "enrich": { "policy_name": "brand-lookup", "field": "brand_id", "target_field": "brand" } }
  ]
}

7. Error handling: using on_failure and ignore_failure deliberately

If a processor fails on a single document, for example because an expected field is missing or a conversion attempt does not work out, the entire pipeline aborts for that document without further configuration and the write fails. Setting ignore_failure: true at the processor level lets a single, non-critical processor be skipped while the rest of the pipeline continues normally, useful for example for an enrich step whose absence would not make the document unusable.

For more critical cases, on_failure, defined at either the processor or the pipeline level, specifies an alternative processor chain that runs when something fails. A proven practice is to tag the document with the error details from _ingest.on_failure_message inside the on_failure block and route it to a separate error index instead of silently dropping it. That way faulty raw data stays visible and can be reprocessed deliberately.

8. Testing pipelines safely with the simulate API

Before a pipeline gets applied to real writes in production, it can be tested against sample documents through POST _ingest/pipeline/_simulate without anything actually landing in the index. Either the full pipeline definition is passed inline, or the name of an already stored pipeline together with a list of test documents that should cover the expected raw data formats, including deliberately malformed examples.

With the verbose=true parameter, the response shows not just the final result but the document state after every individual processor, which helps considerably when debugging multi-step pipelines. A typical workflow is to check new or changed processors exclusively through the simulate API against a set of representative test documents before the pipeline is even attached to a real index.

POST _ingest/pipeline/product-normalize/_simulate?verbose=true
{
  "docs": [
    { "_source": { "sku": " ABC-123 ", "price_raw": "19,90", "categories_raw": "Shoes,Sale" } }
  ]
}

9. Performance implications: when the application side remains the better choice

Ingest processing runs synchronously on the node accepting the write, which puts it directly into that node's CPU budget and the bulk request's latency. An expensive grok expression with many alternatives, or a computation-heavy script processor applied to every single document, can become a noticeable bottleneck at high indexing throughput, especially during an initial bulk load of a large catalog.

As a rule of thumb, ingest pipelines fit best for structural normalization close to the target schema, for simple field renames, type conversions, and enrichment from stable reference data. Computation-heavy, business-logic-laden transformations that already exist in application code or have complex external dependencies are often better kept on the application side, where they can scale independently of indexing throughput and be tested separately.

10. Summary

11. FAQ: Ingest Pipelines: The Essentials at a Glance