Ingest pipelines & processors

This section documents the current Elasticsearch 9.x line (with 8.19 as the final 8.x release) as published at the Elasticsearch documentation, which is the reference these pages are written and verified against. No specific patch version is pinned. Some capabilities (Kibana-only UIs, the ML/NLP model-management workflow, cross-cluster replication, and parts of the paid / serverless-only surface) are linked, not documented in depth.

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production, as Elasticsearch iterates quickly.

This section’s bibliography lists the reference material consulted while preparing these pages.

An ingest pipeline is an ordered list of processors that Elasticsearch applies to each document just before it is written to the index. It covers the light transforms — renaming fields, parsing a log line, looking up a geolocation, deriving a value — that would otherwise need a separate stream-processing tier. This page shows how to create and attach a pipeline, the processors you reach for most, how to handle processor failures, and how to iterate with the _simulate API.

Where a pipeline runs

Every indexing request — a single index, update, or bulk operation, or a reindex — can name a pipeline. When it does, the coordinating node forwards the document to a node with the ingest role, which runs the processors in order and then hands the transformed document to the normal indexing path. By default every node has the ingest role; on a larger cluster you give a subset of nodes that role and keep heavy enrichment off the data nodes.

The pipeline sees the document before mapping is applied, so a processor can create the exact field shapes your mapping expects (parsing a string into a geo_point, coercing a string to long, splitting a message into typed sub-fields). It cannot see other documents except through an enrich processor. See Ingest pipelines for the model.

Creating and attaching a pipeline

Create with PUT /_ingest/pipeline/<id>

A pipeline is a cluster-state object identified by an id. PUT creates or replaces it wholesale.

PUT /_ingest/pipeline/weblogs
{
  "description": "Parse and enrich raw web log lines",
  "processors": [
    { "set":     { "field": "received_at", "value": "{{_ingest.timestamp}}" } },
    { "convert": { "field": "http.response.status_code", "type": "integer" } },
    { "rename":  { "field": "clientip", "target_field": "source.ip", "ignore_missing": true } }
  ]
}

GET /_ingest/pipeline/weblogs
DELETE /_ingest/pipeline/weblogs

See Create or update pipeline API and Get pipeline API.

Attach per request with ?pipeline=

Name the pipeline in the query string of any write. This is explicit and wins for one-off backfills.

POST /weblogs/_doc?pipeline=weblogs
{ "clientip": "203.0.113.7", "http": { "response": { "status_code": "200" } } }

# Bulk: set it once for the whole request...
POST /_bulk?pipeline=weblogs
{ "index": { "_index": "weblogs" } }
{ "clientip": "203.0.113.8", "http": { "response": { "status_code": "404" } } }

# ...or per action line, which overrides the request-level one.
POST /_bulk
{ "index": { "_index": "weblogs", "pipeline": "weblogs" } }
{ "clientip": "203.0.113.9" }

# Reindex through a pipeline.
POST /_reindex
{ "source": { "index": "weblogs-raw" }, "dest": { "index": "weblogs", "pipeline": "weblogs" } }

Attach by default: index.default_pipeline and index.final_pipeline

Two index settings run a pipeline automatically so callers do not have to pass ?pipeline=. Set them on the index, or on an index template so every new backing index inherits them.

PUT /weblogs/_settings
{
  "index.default_pipeline": "weblogs",
  "index.final_pipeline": "add-ingest-metadata"
}

  • index.default_pipeline runs when the request does not name one; passing ?pipeline=_none skips it.

  • index.final_pipeline always runs, after the request pipeline and the default pipeline, and cannot be bypassed — use it for invariants such as stamping an ingest-time field on every document.

See Set a default pipeline.

Common processors

Each entry in processors is an object keyed by the processor name. The full catalogue is Ingest processor reference; the ones below cover most pipelines.

Field shaping: set, remove, rename, convert

PUT /_ingest/pipeline/shape-fields
{
  "processors": [
    { "set":     { "field": "env", "value": "prod", "override": false } },
    { "set":     { "field": "full_name", "value": "{{first}} {{last}}" } },
    { "rename":  { "field": "ts", "target_field": "@timestamp", "ignore_missing": true } },
    { "convert": { "field": "price", "type": "double", "ignore_missing": true } },
    { "remove":  { "field": ["debug", "internal_note"], "ignore_missing": true } }
  ]
}

set writes a literal or a Mustache-style \{{ field }} template value; rename moves a field; convert changes its type (integer, long, float, double, boolean, ip, string, auto); remove drops fields. set, rename, remove, gsub, lowercase, uppercase, trim, split, and join are the small string/field tools you combine freely. See set, rename, convert, and remove.

Parsing text: grok and dissect

grok matches a line against named regular-expression patterns and extracts typed fields; dissect splits on fixed delimiters with no regex, which is faster when the structure is rigid.

PUT /_ingest/pipeline/parse-line
{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{IPORHOST:source.ip} %{USER:auth.user} \\[%{HTTPDATE:ts}\\] \"%{WORD:http.method} %{DATA:url.path}\" %{NUMBER:http.status:int}"]
      }
    },
    {
      "dissect": {
        "field": "syslog_line",
        "pattern": "%{ts} %{+ts} %{host} %{proc}: %{msg}"
      }
    }
  ]
}

See grok processor (and its GET /_ingest/processor/grok pattern list) and dissect processor.

date

date parses one or more date formats from a source field into a target field (@timestamp by default), which is what turns the string grok extracted into a real date for the mapping.

PUT /_ingest/pipeline/parse-timestamp
{
  "processors": [
    {
      "date": {
        "field": "ts",
        "target_field": "@timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z", "ISO8601"],
        "timezone": "UTC"
      }
    }
  ]
}

See date processor.

geoip

geoip looks up an IP address in a bundled MaxMind database and adds city_name, country_iso_code, location (a geo_point), and more. Databases refresh automatically via the GeoIP downloader.

PUT /_ingest/pipeline/geo-enrich
{
  "processors": [
    { "geoip": { "field": "source.ip", "target_field": "source.geo", "ignore_missing": true } }
  ]
}

See geoip processor; Geospatial covers querying the resulting geo_point.

script

script runs a Painless script for logic no declarative processor covers — conditional fields, arithmetic, reshaping. The document is ctx.

PUT /_ingest/pipeline/derive
{
  "processors": [
    {
      "script": {
        "lang": "painless",
        "source": "ctx.bytes_kb = ctx.bytes != null ? Math.round(ctx.bytes / 1024.0) : 0; if (ctx.http?.status >= 500) { ctx.level = 'error' }"
      }
    }
  ]
}

See script processor; Query languages & scripting covers Painless itself.

enrich (and the enrich policy)

enrich joins incoming documents against a separate index at ingest time — the ingest-side answer to a lookup table. You define an enrich policy naming the source index, the field to match on, and the fields to copy, then execute the policy to compile it into an internal system index. Re-execute it whenever the source data changes.

# 1. Reference data lives in an ordinary index.
PUT /users-ref/_doc/u-42
{ "email": "ada@example.com", "department": "R&D", "tier": "gold" }

# 2. Define, then execute, an enrich policy.
PUT /_enrich/policy/user-lookup
{
  "match": {
    "indices": "users-ref",
    "match_field": "email",
    "enrich_fields": ["department", "tier"]
  }
}
POST /_enrich/policy/user-lookup/_execute

# 3. Use it in a pipeline.
PUT /_ingest/pipeline/attach-user
{
  "processors": [
    {
      "enrich": {
        "policy_name": "user-lookup",
        "field": "email",
        "target_field": "user",
        "ignore_missing": true
      }
    }
  ]
}

See Enrich your data, enrich processor, and Create enrich policy API. For query-time joins instead, see Joins & relationships.

pipeline (nested pipelines)

The pipeline processor calls another pipeline, so shared steps live in one place and are composed.

PUT /_ingest/pipeline/common-metadata
{ "processors": [ { "set": { "field": "ingest.version", "value": "2" } } ] }

PUT /_ingest/pipeline/weblogs-v2
{
  "processors": [
    { "pipeline": { "name": "common-metadata" } },
    { "grok": { "field": "message", "patterns": ["%{COMBINEDAPACHELOG}"] } }
  ]
}

See pipeline processor.

Handling failures: on_failure and ignore_failure

By default, a processor that throws aborts the whole pipeline and the document is rejected. Two knobs change that:

  • ignore_failure: true on a processor swallows its error and moves to the next processor.

  • on_failure (on a single processor, or on the pipeline as a whole) is a fallback list of processors that runs when something fails; _ingest.on_failure_message holds the error text.

PUT /_ingest/pipeline/robust
{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{IP:source.ip} %{GREEDYDATA:rest}"],
        "on_failure": [
          { "set": { "field": "event.parse_error", "value": "grok failed" } }
        ]
      }
    },
    { "convert": { "field": "bytes", "type": "long", "ignore_failure": true } }
  ],
  "on_failure": [
    { "set":  { "field": "_index", "value": "weblogs-dead-letter" } },
    { "set":  { "field": "error.message", "value": "{{_ingest.on_failure_message}}" } }
  ]
}

Routing failures to a dead-letter index (by overwriting _index in on_failure) keeps a bad document from blocking a bulk request while preserving it for inspection. See Handling pipeline failures.

Testing with the _simulate API

POST /_ingest/pipeline/<id>/_simulate runs a stored pipeline against sample documents without indexing anything. Post an inline pipeline object instead of an id to try changes before saving them, and add ?verbose to see the document after each processor.

POST /_ingest/pipeline/weblogs/_simulate
{
  "docs": [
    { "_source": { "clientip": "203.0.113.7", "http": { "response": { "status_code": "200" } } } }
  ]
}

# Iterate on a not-yet-saved pipeline, step by step.
POST /_ingest/pipeline/_simulate?verbose
{
  "pipeline": {
    "processors": [
      { "rename":  { "field": "clientip", "target_field": "source.ip" } },
      { "convert": { "field": "http.response.status_code", "type": "integer" } }
    ]
  },
  "docs": [
    { "_source": { "clientip": "203.0.113.7", "http": { "response": { "status_code": "200" } } } }
  ]
}

See Simulate pipeline API. There is also a broader Simulate ingest API that exercises the default/final pipelines and mapping together.

Ingest pipelines are for per-document transforms that run inside Elasticsearch. Aggregating across events, joining large streams, buffering, or reading from queues and files belongs upstream — Logstash and the Elastic Agent / Beats processors cover those heavier jobs and can themselves target a pipeline on the way in.

Continue with Indexing, CRUD & bulk for the write operations pipelines hook into, or Text analysis for how the resulting fields are tokenized at index time.