Learn Elasticsearch - Ingest Pipelines & Data Preprocessing

#Introduction

Raw data in the real world is rarely tidy. Server logs arrive as one long string, date fields are free-form text, IP addresses come without location, and some data needs to be added before indexing. Forcing the application to handle all of this means complicated sender code — when the solution can live in one place: inside Elasticsearch itself.

Ingest pipelines define the preprocessing steps that run automatically as documents come in. Episode 12 covers the common processors, conditional processing and pipeline chaining, parsing with grok, the enrich processor, GeoIP, the user agent parser, and the script processor.

#Ingest Node and Processors

Every node with the ingest role can run pipelines. A pipeline is a list of processors executed in sequence; each processor receives a document and may modify it. If one processor fails, the document is rejected — debug with _ingest/pipeline/_simulate.

Pipelines are created with PUT /_ingest/pipeline/nama, then attached when indexing documents:

{
  "processors": [
    { "set": { "field": "environment", "value": "production" } },
    { "remove": { "field": "sensitive_token" } }
  ]
}

Apply it via PUT /logs/_doc/1?pipeline=normalisasi-log. With this pattern, the application just sends raw documents and the pipeline ensures consistency — all services use the same rules.

#Common Processors

{
  "processors": [
    {
      "convert": {
        "field": "http_status",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "rename": {
        "field": "addr",
        "target_field": "source_ip"
      }
    }
  ]
}

Note ignore_missing: if the field doesn't exist, the processor doesn't error — important for data with optional fields.

#Conditional Processing and Pipeline Chaining

Not all documents need the same treatment. A processor can be given a condition with simple Painless syntax:

{ "set": { "field": "severity_code", "value": 5, "if": "ctx.level == 'error'" } }

Pipelines can also call other pipelines (chaining) — breaking large preprocessing into reusable modules:

{
  "processors": [
    { "pipeline": { "name": "normalisasi-field" } },
    { "pipeline": { "name": "enrich-geolokasi" } }
  ]
}

#Parsing Logs with Grok

Grok parses semi-structured text into structured fields using named regex patterns. The most classic example — Apache/nginx access logs:

{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:method} %{DATA:uri} %{NOTSPACE:protocol}\" %{NUMBER:status:int} %{NUMBER:bytes:int}"
        ]
      }
    }
  ]
}

A single log line 127.0.0.1 - - [03/Aug/2026:10:00:00 +0700] "GET /produk HTTP/1.1" 200 512 immediately turns into the structured fields client_ip, method, uri, status, bytes. Patterns like %{IPORHOST}, %{HTTPDATE}, %{WORD} are built-in — you don't need to write regex from scratch.

#Enrich, GeoIP, and User Agent

The enrich processor joins incoming documents with data from another index. For example: transaction logs only carry product_id; you want to add the product name and category from a reference index. Enrich requires an enrich policy that configures the source index, then it gets executed:

PUT /_enrich/policy/produk-policy
POST /_enrich/policy/produk-policy/_execute

{
  "processors": [
    {
      "enrich": {
        "policy_name": "produk-policy",
        "field": "product_id",
        "target_field": "produk_detail"
      }
    }
  ]
}

The result: a produk_detail field appears containing name and category — data enriched without changing the sending application.

The GeoIP processor maps an IP address to a geographic location; the user agent processor parses a User-Agent string into structured fields:

{
  "processors": [
    { "geoip": { "field": "source_ip", "target_field": "geo" } },
    { "user_agent": { "field": "ua_string", "target_field": "ua" } }
  ]
}

Documents automatically get geo.country_name, geo.location (geo_point type) for map visualizations, as well as ua.name, ua.os.name, ua.device.type for device analysis.

#Script Processor

Sometimes the transformation logic is too specific for a built-in processor. The script processor uses the Painless language for free-form transformations:

{ "script": { "source": "ctx.total_harga = (ctx.harga_satuan ?: 0) * (ctx.jumlah ?: 0); ctx.nama.trim()" } }

Painless is Elasticsearch's specialized scripting language, safe to run on the server. For simple logic, use built-in processors; save scripts for cases that genuinely need flexibility.

#Common Mistakes

1. Grok pattern doesn't match real data. Always test with _ingest/pipeline/_simulate using real log samples.
2. Forgetting ignore_missing. Optional fields that don't exist make the processor error.
3. Enrich policy not executed. A policy that hasn't been run has no enrich index yet.
4. Pipeline attached to a single index only. Attach the pipeline to a template (episode 4) for consistency.
5. Unbounded nested pipelines. Chaining that's too deep is hard to debug — limit the levels.

#Conclusion

In episode 12 you mastered ingest pipelines: common processors (set, remove, rename, convert), conditional processing with Painless, pipeline chaining, log parsing with grok, the enrich processor, GeoIP and user agent parsers, and the script processor for custom logic.

Key takeaways:

• A pipeline is a sequence of preprocessing steps that run as documents arrive.
• Grok turns log strings into structured fields with built-in patterns.
• Enrich joins data from a reference index; don't forget to execute the policy.
• GeoIP and user agent add location and device dimensions to data.
• Attach pipelines via templates so they're consistent across all indexes.
• Test with _ingest/pipeline/_simulate before using in production.

Often the data that's already indexed needs to change — wrong mapping, changed format, or a need to merge. In episode 13 we'll cover reindex and update by query: when and why to reindex, reindexing from a remote cluster, update by query for mass updates, handling conflicts, throttling, and delete by query. See you there!