08c — Ingest pipelines and best practices

The final part of the course closes the loop: how to transform data cleanly at index time using ES-native ingest pipelines, when to reach for Beats or Logstash instead, and a concise checklist of everything covered across both days.

Goals for this part

Build an ingest pipeline with common processors and test it with _simulate
Set on_failure handlers so bad documents are visible rather than silently lost
Choose the right ingestion tool for a given scenario (ingest pipeline vs Beats vs Logstash)
Apply the course best-practices checklist when designing a new integration

8.4 Ingest pipelines and Logstash/Beats — when to use what

Ingest pipelines — ES-native, runs on ingest nodes

Ingest pipelines process documents inside Elasticsearch before they are written to shards. They run on ingest-role nodes (or any node in small clusters).

Common processors:

Processor	What it does
`grok`	Pattern extraction from unstructured text (log lines)
`dissect`	Faster delimiter-based field splitting (prefer over grok when possible)
`date`	Parse a timestamp string into a date field
`convert`	String → long, double, boolean, etc.
`rename`	Rename a field
`remove`	Drop unwanted fields
`set`	Set a field value (constant or from another field)
`lowercase` / `uppercase`	Normalize string case
`geoip`	Enrich an IP field with city/country/coordinates
`user_agent`	Parse a user-agent string into structured fields
`script`	Arbitrary Painless logic (powerful; use sparingly)

Register a pipeline

PUT /_ingest/pipeline/normalize-orders-v1
{
  "description": "Normalize shop-orders-demo documents on ingest",
  "processors": [
    {
      "set": {
        "field": "@timestamp",
        "value": "{{created_at}}",
        "ignore_failure": true
      }
    },
    {
      "date": {
        "field": "@timestamp",
        "formats": ["ISO8601", "yyyy-MM-dd'T'HH:mm:ssZ"],
        "ignore_failure": true
      }
    },
    {
      "lowercase": {
        "field": "customer.name",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "amount",
        "type": "double",
        "ignore_missing": true,
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": ["internal_debug_field"],
        "ignore_missing": true
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "ingest.error",
        "value": "{{_ingest.on_failure_message}}"
      }
    }
  ]
}

Test with `_simulate` before deploying

POST /_ingest/pipeline/normalize-orders-v1/_simulate
{
  "docs": [
    {
      "_index": "shop-orders-demo",
      "_source": {
        "created_at": "2026-05-30T09:15:00Z",
        "customer": { "name": "Alice SMITH" },
        "amount": "149.99",
        "internal_debug_field": "drop-me"
      }
    }
  ]
}

Inspect the _simulate response’s doc._source to verify the transformation before attaching it to an index.

Attach to an index

Per-request (one-off or testing):

POST /shop-orders-demo/_doc?pipeline=normalize-orders-v1
{ "created_at": "2026-05-30T09:15:00Z", "amount": "149.99" }

As the index default (applies to all documents indexed to that index):

PUT /shop-orders-demo/_settings
{
  "index.default_pipeline": "normalize-orders-v1"
}

Decision guide: pipeline vs Beats vs Logstash

Scenario	Best choice
App indexes directly into ES; light transformation needed (normalize, convert, enrich)	Ingest pipeline — ES-native, zero extra infrastructure
Collecting logs/metrics from servers, containers, or services; minimal transformation	Beats (Filebeat, Metricbeat) — lightweight agents, designed to ship and hand off
Complex multi-step parsing, multiple input sources or sinks, buffering, fan-out	Logstash — full pipeline engine with persistent queues
Heavy enrichment that would be expensive on ingest nodes (machine learning, heavy scripts)	Logstash or pre-processing before indexing

Comparison table:

	Ingest pipeline	Beats	Logstash
Runs inside	Elasticsearch	Agent on source host	Separate JVM process
Infrastructure	None extra	Lightweight agents	Dedicated nodes/pods
Transformation power	Medium (processors, scripts)	Minimal (some modules)	High (full plugin ecosystem)
Buffering / durability	None (synchronous)	Limited	Persistent queues
Multi-destination output	No	No (sends to ES or Logstash)	Yes
Typical use	Enrich/normalize at index time	Collect and ship	Complex ETL, fan-out

Use ingest pipelines by default for anything you can express in processors. Add Logstash only when you genuinely need its power — it adds operational complexity. Beats are for collection, not transformation.

8.5 Best practices wrap-up

A concise checklist tying together the entire course. Work through it when designing a new index, reviewing an integration, or doing a health check on an existing cluster.

Mappings

Use explicit mappings for every critical field — do not rely on dynamic detection for production indices.
Apply multi-fields (text + .raw keyword) for any field you need to both search full-text and aggregate/sort.
Set dynamic: strict (reject unexpected fields) or dynamic: false (ignore them) on indices where schema drift is a risk.
Use flattened for arbitrary key/value blobs (labels, metadata, user-supplied JSON) to avoid field explosion.
Use nested only when per-item correctness in arrays of objects is required — it has query cost.
Define mappings in index templates so every new rollover index inherits the correct schema.

Queries

Put non-scoring conditions in filter context (bool.filter) — they are cached and faster.
Always scope to a time range on time-series indices; “query all history” will be expensive.
Avoid high-cardinality terms aggregations on unbounded fields (IDs, UUIDs) — use composite for paging buckets if you need enumeration.
Use search_after for deep pagination; avoid from/size beyond a few thousand results.
Profile expensive queries in Dev Tools with "profile": true before running them in production.

Shards, ILM, and templates

Target 10–50 GB per shard; stay under a few hundred shards per node.
Use ILM with rollover for time-series indices — it keeps shards predictably sized over time.
Use write aliases as the indexing endpoint so rollover is transparent to the application.
Do not create one index per tenant or per day without a retention plan — shard count creep is a real stability risk.

Aliases for zero-downtime changes

When you need to fix a mapping or reindex: 1. Create the new index with the correct mapping. 2. Reindex from old to new. 3. Flip the alias atomically to point to the new index. 4. Applications pointing at the alias see no interruption.

POST /_aliases
{
  "actions": [
    { "remove": { "index": "shop-orders-demo-v1", "alias": "shop-orders" } },
    { "add":    { "index": "shop-orders-demo-v2", "alias": "shop-orders" } }
  ]
}

Elasticsearch as a derived store

Know what is the source of truth for your data. If it is a database or event stream, ES is a derived index — you can rebuild it.
If ES is the only store for some data, enforce snapshot discipline and test restores.
Do not use ES as a primary transactional store — it is optimized for search, not ACID writes.

Integration hygiene

Use ingest pipelines for light normalization; test every pipeline with _simulate before deploying.
Set on_failure handlers in pipelines so bad documents are marked with an error field rather than silently dropped.
Use index.default_pipeline in templates so the pipeline is applied automatically.
Validate your bulk indexing error responses — the bulk API returns 200 even when individual documents fail.

# Check for per-document errors in a bulk response
# errors: true in the response body means at least one item failed
# Always check "items[].index.error" when errors is true

Monitor heap, disk, and rejections

Heap > 75–85% sustained: investigate. Reduce cardinality, shard count, or query concurrency.
Disk usage rising toward 85%: fix retention or add capacity before the watermark hits.
Any thread pool rejections: act immediately — requests are being dropped.
Check _cluster/pending_tasks when anything feels slow — a growing queue signals master pressure.

Safety habits

Take a snapshot before any reindex, bulk delete, or major mapping change.
Test your restore process — a snapshot you have never restored is unproven.
Use _cluster/allocation/explain before force-allocating or deleting indices during a health incident.
Never delete indices under pressure without confirming they are not the only copy of that data.

Lab — Build and test an ingest pipeline

Goal: Create a real pipeline, test it with _simulate, attach it to an index, and verify it runs automatically.

Step 1 — Create the pipeline normalize-orders-v1 with these processors: - set to copy created_at into @timestamp - date to parse @timestamp into a proper date type - lowercase on customer.name - convert to ensure amount is double - remove to drop internal_debug_field - on_failure handler to record errors in ingest.error

PUT /_ingest/pipeline/normalize-orders-v1
{
  "description": "Normalize order documents on ingest",
  "processors": [
    {
      "set": {
        "field": "@timestamp",
        "value": "{{created_at}}",
        "ignore_failure": true
      }
    },
    {
      "date": {
        "field": "@timestamp",
        "formats": ["ISO8601"],
        "ignore_failure": true
      }
    },
    {
      "lowercase": {
        "field": "customer.name",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "amount",
        "type": "double",
        "ignore_missing": true,
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": ["internal_debug_field"],
        "ignore_missing": true
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "ingest.error",
        "value": "{{_ingest.on_failure_message}}"
      }
    }
  ]
}

Step 2 — Test with _simulate before touching any index. Use a sample document that has both valid and edge-case fields:

POST /_ingest/pipeline/normalize-orders-v1/_simulate
{
  "docs": [
    {
      "_index": "orders-test",
      "_source": {
        "created_at": "2026-05-30T09:15:00Z",
        "customer": { "name": "Alice SMITH" },
        "amount": "149.99",
        "internal_debug_field": "drop-me"
      }
    },
    {
      "_index": "orders-test",
      "_source": {
        "created_at": "2026-05-30T10:00:00Z",
        "customer": { "name": "BOB JONES" },
        "amount": "not-a-number",
        "internal_debug_field": "also-drop-me"
      }
    }
  ]
}

Questions: 1. For document 1: is @timestamp a proper date? Is customer.name lowercased? Is amount a number? Is internal_debug_field gone? 2. For document 2: what happened when amount could not be converted to double? Did it fail silently (because of ignore_failure: true), or is there an error field?

Step 3 — Test the on_failure handler. Temporarily break the pipeline by removing ignore_failure: true from the convert processor, simulate again with document 2, and observe the ingest.error field:

PUT /_ingest/pipeline/normalize-orders-v1-strict
{
  "description": "Strict version to test on_failure",
  "processors": [
    {
      "convert": {
        "field": "amount",
        "type": "double"
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "ingest.error",
        "value": "{{_ingest.on_failure_message}}"
      }
    }
  ]
}

POST /_ingest/pipeline/normalize-orders-v1-strict/_simulate
{
  "docs": [
    {
      "_index": "orders-test",
      "_source": {
        "amount": "not-a-number"
      }
    }
  ]
}

Is ingest.error populated in the simulated output?

Step 4 — Create a test index and attach the pipeline as index.default_pipeline:

PUT /orders-pipeline-test
{
  "settings": {
    "number_of_replicas": 0,
    "index.default_pipeline": "normalize-orders-v1"
  },
  "mappings": {
    "properties": {
      "@timestamp":   { "type": "date" },
      "created_at":   { "type": "keyword" },
      "customer":     { "type": "object" },
      "amount":       { "type": "double" },
      "ingest":       { "type": "object", "dynamic": true }
    }
  }
}

Step 5 — Index a document WITHOUT specifying the pipeline explicitly — the default pipeline should run automatically:

POST /orders-pipeline-test/_doc
{
  "created_at": "2026-05-30T11:00:00Z",
  "customer": { "name": "CAROL NOVAK" },
  "amount": "299.50",
  "internal_debug_field": "should-be-dropped"
}

Retrieve the document and verify:

GET /orders-pipeline-test/_search

Questions: 1. Is @timestamp present and a proper date? 2. Is customer.name lowercase (“carol novak”)? 3. Is amount stored as a number (not a string)? 4. Is internal_debug_field absent from _source?

Step 6 — Index a document that will trigger on_failure (set amount to a non-numeric string). Verify it is indexed with ingest.error set:

POST /orders-pipeline-test/_doc
{
  "created_at": "2026-05-30T11:05:00Z",
  "customer": { "name": "DAVE TEST" },
  "amount": "broken-value",
  "internal_debug_field": "drop-me"
}

Note: the convert processor has ignore_failure: true, so it will NOT trigger on_failure — the document will be indexed with amount left as a string. To see on_failure in action with a real index, you would need to remove ignore_failure and accept that the conversion failure triggers the error handler. Explain to your classmates: what is the trade-off between ignore_failure: true (never blocks, but silently wrong types) and on_failure (blocked, but always visible)?

Step 7 — Clean up:

DELETE /orders-pipeline-test
DELETE /_ingest/pipeline/normalize-orders-v1
DELETE /_ingest/pipeline/normalize-orders-v1-strict

Takeaway: Always _simulate before deploying a pipeline — it shows the exact transformation without touching any real index. on_failure handlers ensure bad documents surface as searchable error records rather than being silently rejected or silently wrong. Attach pipelines via index.default_pipeline in index templates so the pipeline runs automatically for all new indices in the pattern — no per-request pipeline parameter needed.