Ingest pipelines in Elasticsearch

Introduction

Ingest pipelines help you prepare and change data before saving it in Elasticsearch. They make data clean and ready to use.

You want to add or change fields in your data automatically before storing it.

You need to remove or fix wrong data before it goes into Elasticsearch.

You want to convert data formats, like dates or numbers, during data input.

You want to enrich data by adding information from other sources before saving.

You want to log or track data changes as it enters Elasticsearch.

Syntax

Elasticsearch

{
  "description": "optional description",
  "processors": [
    {
      "processor_type": {
        "field": "field_name",
        "target_field": "new_field_name",
        "other_options": "value"
      }
    },
    {
      "another_processor": {
        "field": "field_name",
        "options": "value"
      }
    }
  ]
}

Processors run in order, one after another.

You can use many types of processors like set, rename, remove, convert, and more.

Examples

This pipeline adds a new field called new_field with the value "hello" to each document.

Elasticsearch

{
  "description": "Add a new field",
  "processors": [
    {
      "set": {
        "field": "new_field",
        "value": "hello"
      }
    }
  ]
}

This pipeline changes the name of a field from old_name to new_name.

Elasticsearch

{
  "processors": [
    {
      "rename": {
        "field": "old_name",
        "target_field": "new_name"
      }
    }
  ]
}

This pipeline converts the age field to an integer type.

Elasticsearch

{
  "processors": [
    {
      "convert": {
        "field": "age",
        "type": "integer"
      }
    }
  ]
}

Sample Program

This example creates a pipeline named sample_pipeline that adds a status field with value "new" and renames the user field to username. Then it simulates running this pipeline on a document.

Elasticsearch

PUT _ingest/pipeline/sample_pipeline
{
  "description": "Pipeline to add and rename fields",
  "processors": [
    {
      "set": {
        "field": "status",
        "value": "new"
      }
    },
    {
      "rename": {
        "field": "user",
        "target_field": "username"
      }
    }
  ]
}

POST _ingest/pipeline/sample_pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "user": "alice",
        "message": "Hello world"
      }
    }
  ]
}

Important Notes

Use the _simulate API to test pipelines without indexing data.

Processors run in the order you list them, so plan carefully.

If a processor fails, the whole pipeline stops unless you handle errors.

Summary

Ingest pipelines let you change data before saving it in Elasticsearch.

You create pipelines with processors that do tasks like add, rename, or convert fields.

Test pipelines with the simulate API to see changes before indexing.

Practice

1. What is the main purpose of an ingest pipeline in Elasticsearch?

easy

A. To create visualizations from data

B. To query documents faster

C. To backup Elasticsearch data

D. To modify documents before they are indexed

Solution

1. Step 1: Understand ingest pipeline role

   An ingest pipeline processes documents before they are stored in Elasticsearch.

2. Step 2: Compare options to pipeline purpose

   Only To modify documents before they are indexed matches this role; others describe unrelated tasks.

3. Final Answer:

   To modify documents before they are indexed -> Option D

4. Quick Check:

   Ingest pipeline = modify before indexing [OK]

Hint: Ingest pipelines change data before saving [OK]

Common Mistakes:

• Confusing pipelines with query optimization
• Thinking pipelines create backups
• Assuming pipelines generate reports

2. Which of the following is the correct way to define a processor that adds a field "source": "web" in an ingest pipeline?

easy

A. "processors": [{"add": {"field": "source", "value": "web"}}]

B. "processors": [{"remove": {"field": "source", "value": "web"}}]

C. "processors": [{"set": {"field": "source", "value": "web"}}]

D. "processors": [{"append": {"field": "source", "value": "web"}}]

Solution

1. Step 1: Identify processor to add or set field

   The set processor assigns or adds a field with a value.

2. Step 2: Check syntax correctness

   "processors": [{"set": {"field": "source", "value": "web"}}] uses set with correct JSON structure; add is not a valid processor name.

3. Final Answer:

   "processors": [{"set": {"field": "source", "value": "web"}}] -> Option C

4. Quick Check:

   Use set to add fields [OK]

Hint: Use 'set' processor to add or update fields [OK]

Common Mistakes:

• Using 'add' instead of 'set' processor
• Confusing 'append' with adding new fields
• Trying to remove field when adding

3. Given this ingest pipeline snippet:

{"processors": [{"rename": {"field": "old_name", "target_field": "new_name"}}]}

What happens to a document {"old_name": "value"} when processed?

medium

A. {"old_name": "value", "new_name": "value"}

B. {"new_name": "value"}

C. {"old_name": null}

D. {"old_name": "value"}

Solution

1. Step 1: Understand rename processor effect

   The rename processor moves the value from old_name to new_name, removing the original field.

2. Step 2: Apply to example document

   Input {"old_name": "value"} becomes {"new_name": "value"} after processing.

3. Final Answer:

   {"new_name": "value"} -> Option B

4. Quick Check:

   Rename moves field, removes old [OK]

Hint: Rename moves field value, deletes old field [OK]

Common Mistakes:

• Thinking rename duplicates field
• Assuming old field remains
• Confusing rename with copy

4. You wrote this pipeline processor:

{"processors": [{"convert": {"field": "age", "type": "integer"}}]}

But when simulating, you get an error. What is the likely cause?

medium

A. The field "age" does not exist in the document

B. The "convert" processor does not exist

C. The "type" value "integer" is invalid

D. The pipeline must have a "description" field

Solution

1. Step 1: Check processor and type validity

   The "convert" processor and type "integer" are valid in Elasticsearch ingest pipelines.

2. Step 2: Identify common error cause

   If the field "age" is missing in the document, conversion fails causing error during simulation.

3. Final Answer:

   The field "age" does not exist in the document -> Option A

4. Quick Check:

   Missing field causes convert error [OK]

Hint: Convert errors if field missing in document [OK]

Common Mistakes:

• Assuming convert processor is invalid
• Thinking 'integer' is wrong type
• Believing description is mandatory

5. You want to create an ingest pipeline that adds a field "status": "new" only if the field "priority" exists. Which processor setup achieves this?

hard

A. "processors": [{"set": {"field": "status", "value": "new", "if": "ctx.priority != null"}}]

B. "processors": [{"set": {"field": "status", "value": "new"}}, {"exists": {"field": "priority"}}]

C. "processors": [{"conditional": {"field": "priority", "processor": {"set": {"field": "status", "value": "new"}}}}]

D. "processors": [{"set": {"field": "status", "value": "new"}, "condition": "priority"}]

Solution

1. Step 1: Understand conditional execution in pipelines

   Processors support an "if" condition using painless scripts to run only when true.

2. Step 2: Identify correct syntax for conditional set

   "processors": [{"set": {"field": "status", "value": "new", "if": "ctx.priority != null"}}] uses "if": "ctx.priority != null" which runs set only if "priority" exists.

3. Final Answer:

   "processors": [{"set": {"field": "status", "value": "new", "if": "ctx.priority != null"}}] -> Option A

4. Quick Check:

   Use "if" with painless script for conditions [OK]

Hint: Use 'if' with painless script to conditionally run processor [OK]

Common Mistakes:

• Using non-existent 'conditional' processor
• Placing condition outside processor block
• Misusing 'exists' as processor