Ingest pipelines: Elasticsearch best practices explained
When documents land in Elasticsearch, they rarely arrive in the shape you need them. Field names might be inconsistent, timestamps in unexpected formats, or critical context missing altogether. Ingest pipelines let you transform, enrich and validate documents before they hit your indices—keeping your indexing logic clean and your data reliable. This is Day 45 of our Elasticsearch & Search Engineering series, and today we're diving into the processors, enrichment strategies, and patterns that production teams use to build robust pipelines.
Understanding ingest pipelines fundamentally
An ingest pipeline is a series of processors that run on a document before indexing. Each processor transforms the document, extracts fields, drops data, or flags it for special handling. Unlike search-time analyzers, which transform text for matching, ingest processors reshape the document structure itself.
Pipelines run on the ingest node—a dedicated node type that handles this work without competing with search or indexing resources. You define a pipeline once, attach it to one or more indices, and every document flowing through that index gets processed automatically.
The key advantage: your application code doesn't need to know about field mapping quirks, date formats, or enrichment logic. That lives in the pipeline. When requirements change—a new field to extract, a rename to standardize naming—you update the pipeline, not your application.
Processors: the building blocks
Elasticsearch ships with dozens of processors. We'll cover the ones you'll reach for repeatedly in production.
Set processor: add or override fields
The simplest processor adds or overwrites a field with a static value or computed expression.
{
"processors": [
{
"set": {
"field": "pipeline_version",
"value": "2.1"
}
},
{
"set": {
"field": "environment",
"value": "{{source_system}}"
}
}
]
}
This is useful for annotating documents with metadata: which system sent them, when the pipeline ran, or a default category. The second example uses the Mustache template syntax to reference fields already in the document.
Rename processor: standardize field names
Legacy systems often send data with inconsistent field naming. One API uses customer_id, another customerId, and a third cust_id. Rename them all to a single canonical name.
{
"processors": [
{
"rename": {
"field": "customerId",
"target_field": "customer_id"
}
},
{
"rename": {
"field": "cust_id",
"target_field": "customer_id",
"ignore_missing": true
}
}
]
}
The ignore_missing flag prevents the pipeline from failing if a field doesn't exist. This is critical when different documents might have different shapes.
Grok processor: parse unstructured text
Grok is a pattern-matching language for extracting fields from free text. It's invaluable for logs, access records, or any semi-structured content.
{
"processors": [
{
"grok": {
"field": "message",
"patterns": [
"%{IP:client_ip} %{WORD:http_method} %{URIPATH:request_path} %{INT:status_code:int} %{INT:response_time_ms:int}"
]
}
}
]
}
Given a log line like 192.168.1.1 GET /api/users 200 45, grok extracts client_ip, http_method, request_path, status_code, and response_time_ms as distinct searchable fields. The :int suffix casts the value to an integer, which is much better for range queries than storing it as a string.
Date processor: parse and normalize timestamps
Systems send timestamps in ISO 8601, Unix milliseconds, or custom formats. The date processor parses them into a standard format that Elasticsearch understands.
{
"processors": [
{
"date": {
"field": "event_time",
"target_field": "@timestamp",
"formats": [
"ISO8601",
"epoch_millis",
"yyyy-MM-dd HH:mm:ss"
],
"timezone": "UTC"
}
}
]
}
The processor tries each format in order until one matches. By mapping to @timestamp, you ensure Elasticsearch treats it as a time-series event, unlocking time-based retention policies and aggregations.
JSON processor: unwrap nested JSON in strings
Sometimes a field contains JSON as a string rather than as an object. Parse it into proper nested fields.
{
"processors": [
{
"json": {
"field": "metadata",
"target_field": "parsed_metadata"
}
}
]
}
This transforms {"metadata": "{\"user_id\": 123, \"region\": \"us-west\"}"} into a searchable parsed_metadata.user_id field.
Remove and drop processors: filter documents or fields
Remove unwanted fields to reduce index size. Drop entire documents if they don't meet criteria.
{
"processors": [
{
"remove": {
"field": [
"internal_debug",
"temporary_token"
]
}
},
{
"drop": {
"if": "ctx.status_code >= 500"
}
}
]
}
The drop processor with a condition is especially useful: only index non-error logs, or filter to specific customer segments. This reduces storage and improves indexing performance.
Enrich: add data from lookup indices
Enrichment joins document data with reference data stored in separate indices. A common case: your events contain a product_id, but you want to index the product category, price, and supplier region alongside it.
{
"processors": [
{
"enrich": {
"policy_name": "product_enrichment",
"field": "product_id",
"target_field": "product_details",
"override": false
}
}
]
}
Before using the enrich processor, you create an enrich policy that points to a source index:
{
"policy_name": "product_enrichment",
"match": {
"indices": ["products"],
"match_field": "id",
"enrich_fields": ["category", "price", "supplier_region"]
}
}
When an event with product_id: 42 flows through, the processor looks up id: 42 in the products index and merges the matching document's fields into product_details.
Enrichment works best when the reference data is relatively static and the lookup cardinality is reasonable. For high-velocity streams with millions of distinct IDs, consider populating enriched fields at index time in your application instead.
Default pipelines: apply automatically to an index
You can set a pipeline as the default for an index, so every document is automatically processed without the client needing to specify it.
PUT /my-index-logs-2025.01
{
"settings": {
"index.default_pipeline": "log-parser"
},
"mappings": {
"properties": {
"message": {"type": "text"},
"client_ip": {"type": "ip"},
"status_code": {"type": "integer"}
}
}
}
Now any document indexed to this index uses the log-parser pipeline automatically. You can also set a default pipeline at the cluster level, though per-index configuration is usually clearer.
In production, this pattern shines when you have many indices following the same data flow. One pipeline definition, applied consistently across your entire data lake.
Test: simulate pipelines with sample documents
Before rolling a pipeline to production, test it with real data. The simulate API lets you run documents through a pipeline and see the output without actually indexing.
POST _ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"grok": {
"field": "message",
"patterns": ["%{IP:client_ip} %{WORD:http_method} %{URIPATH:request_path} %{INT:status_code:int}"]
}
},
{
"date": {
"field": "timestamp",
"formats": ["ISO8601"]
}
}
]
},
"docs": [
{
"_source": {
"message": "192.168.1.10 GET /api/users 200",
"timestamp": "2025-01-15T10:30:00Z"
}
},
{
"_source": {
"message": "10.0.0.5 POST /api/auth 401",
"timestamp": "2025-01-15T10:31:15Z"
}
}
]
}
The response shows the transformed document structure for each input:
{
"docs": [
{
"doc": {
"_source": {
"message": "192.168.1.10 GET /api/users 200",
"timestamp": "2025-01-15T10:30:00Z",
"client_ip": "192.168.1.10",
"http_method": "GET",
"request_path": "/api/users",
"status_code": 200,
"@timestamp": "2025-01-15T10:30:00.000Z"
}
}
}
]
}
This lets you validate field extraction, type casting, and enrichment before committing changes. We recommend adding simulation tests to your CI/CD pipeline to catch regressions when pipeline logic changes.
Common mistakes
- Ignoring missing fields everywhere: Adding
ignore_missing: trueto every processor hides data quality issues. Use it selectively, and monitor pipeline failures to spot upstream problems. - Over-enriching at scale: Enrichment works per-document and hits the enrich index lookup. At millions of documents per day, enrichment latency adds up. Pre-compute enriched data in your application or batch-enrich offline.
- Grok patterns that are too greedy: Patterns like
.*consume more text than intended. Test patterns thoroughly with the simulate API before deploying. - Storing transformed data in the wrong field: When using processors like json or grok, double-check your
target_field. Accidentally overwriting source data makes debugging much harder. - Not versioning pipelines: Treat pipelines as code. Store them in version control, document changes, and test updates before promotion.
Key takeaways
- Processors let you transform, parse, and enrich documents before indexing—keeping application code simpler and more maintainable.
- Grok and date processors are your allies for parsing logs and timestamps; rename and set processors standardize messy incoming data.
- Enrichment adds reference data from lookup indices, but monitor latency at high scale and consider pre-enrichment alternatives.
- Set a default pipeline on your indices to ensure consistent processing without client coordination.
- Always test pipelines with the simulate API before deploying to production, and treat pipeline definitions as versioned infrastructure code.
- Remove fields you don't need and use the drop processor to filter unwanted documents early, reducing storage and indexing load.
Comments
Loading comments…