Skip to main content
Mapagam

Working with Data Transforms

1. Understanding Transforms

Transform Type Description Use Case Example
Ingest Pipeline Processes documents as they are ingested into Elasticsearch Preprocessing data before indexing 
{
  "description": "Pipeline to enrich logs",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"]
      }
    }
  ]
}
Data Preprocessing Modifies data fields before indexing Normalizing timestamps, parsing JSON 
{
  "script": {
    "source": "ctx.timestamp = ZonedDateTime.ofInstant(Instant.ofEpochMilli(ctx.timestamp), ZoneId.systemDefault()).format(DateTimeFormatter.ISO_INSTANT)"
  }
}
Field Extraction Extracts and formats fields from raw data Splitting log lines into structured fields 
{
  "grok": {
    "field": "message",
    "patterns": ["%{IP:client_ip} %{WORD:method} %{URIPATH:uri}"]
  }
}
Note: Ingest pipelines are processed in the order they are defined, and each processor can modify the document before passing it to the next one.

2. Creating Transform Jobs

Property Description Values
id Unique identifier for the transform job "my-transform-job"
source.index Source index pattern to transform data from "logs-*"
dest.index Destination index to write transformed data to "transformed-logs"
pivot Configuration for pivot-based transforms { "group_by": [ { "field": "user.id", "missing_bucket": true } ], "aggregations": { "avg_duration": { "avg": { "field": "duration" } } } }
settings Optional settings for the transform job { "max_page_search_size": 1000 }

Example: Creating a Transform Job

{
  "id": "user-session-transform",
  "source": {
    "index": "user-logs-*"
  },
  "dest": {
    "index": "user-sessions"
  },
  "pivot": {
    "group_by": [
      {
        "field": "user.id",
        "missing_bucket": true
      }
    ],
    "aggregations": {
      "avg_duration": {
        "avg": {
          "field": "duration"
        }
      }
    }
  },
  "settings": {
    "max_page_search_size": 1000
  }
}
Note: Transform jobs must be created before any data is indexed into the source index.
Warning: Ensure that the destination index has a proper mapping to avoid data loss during transformation.

3. Configuring Pivot Transforms

Property Description Values Notes
id Unique identifier for the transform "pivot-transform" Required field
source Source index or indices to transform "logs-*" Supports wildcards
destination Index where transformed data is stored "pivot-logs" Must be a valid index name
pivot Configuration for pivot transformation {} Contains pivot settings

Example: Basic Pivot Transform Configuration

{
  "id": "pivot-transform",
  "source": "logs-*",
  "destination": "pivot-logs",
  "pivot": {
    "group_by": ["user.id"],
    "agg": {
      "total_requests": {
        "sum": {
          "field": "request.count"
        }
      }
    }
  }
}
Note: The pivot transform is ideal for aggregating data by grouping fields and applying aggregate functions like sum, average, min, max.

4. Using Latest Transforms

Transform Type Description Usage Example Notes
script Executes a script to transform documents 
{
  "script": {
    "source": "ctx.my_field = ctx.my_field.toUpperCase()"
  }
}
 Supports Painless scripting language
date Formats date fields in documents 
{
  "date": {
    "field": "timestamp",
    "formats": ["yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"]
  }
}
 Can handle multiple date formats
rename Renames fields in documents 
{
  "rename": {
    "field": "old_name",
    "target_field": "new_name"
  }
}
 Can be used to move fields
set Sets a field to a specific value 
{
  "set": {
    "field": "status",
    "value": "active"
  }
}
 Can set nested fields using dot notation
remove Removes fields from documents 
{
  "remove": {
    "field": "temp_field"
  }
}
 Useful for cleaning up temporary fields

Example: Applying a Script Transform

{
  "transform": {
    "script": {
      "source": "ctx.user_agent = ctx.user_agent.toLowerCase(); ctx.timestamp = new Date().toISOString();"
    }
  }
}
Note: Transforms are applied in the order they appear in the pipeline. Ensure that dependencies between transforms are handled correctly.
Warning: Script transforms can be computationally expensive. Monitor performance when applying complex transformations.

5. Setting Transform Sync Configuration

Property Description Default Value Notes
transform.sync.enabled Enables or disables the sync configuration false Set to true to enable sync
transform.sync.interval Interval (in seconds) to run the sync 30 Minimum value is 1 second
transform.sync.timeout Timeout (in seconds) for sync operations 60 Used to prevent long-running syncs

Example: Configuring Transform Sync

transform:
  sync:
    enabled: true
    interval: 60
    timeout: 120
Note: The sync configuration is applied at runtime and requires a restart of the transform service to take effect.

6. Configuring Retention Policy

Property Description Default Value Notes
indices Pattern to match indices for retention policy * Supports wildcards like * and ?
max_age Maximum age of indices before deletion 30d Can use units like d, h, m
max_size Maximum size of indices before deletion 50gb Can use units like gb, tb
min_index_number Minimum index number to keep 0 Useful for keeping a minimum number of indices

Example: Setting up a retention policy

{
  "policy": {
    "name": "elk-retention-policy",
    "indices": "logs-*",
    "max_age": "30d",
    "max_size": "50gb"
  }
}
Note: Retention policies are applied at the index level and can be managed via the Elasticsearch indices API or through Kibana's Index Management UI.
Warning: Ensure that the retention policy does not delete indices that are still needed for compliance or analysis.

7. Starting and Stopping Transforms

Operation Description Command Status
Start Transform Initiates a data transform job POST /_transform/{transform_id}/_start NEW
Stop Transform Halts a running data transform job POST /_transform/{transform_id}/_stop NEW

Example: Starting a Transform

POST /_transform/my_transform/_start

{
  "timeout": "20m"
}

Example: Stopping a Transform

POST /_transform/my_transform/_stop

{
  "timeout": "10m"
}
Note: Transforms can be started and stopped multiple times. Stopping a transform does not delete the destination index.

8. Previewing Transform Results

Example: Previewing Transform Results in Kibana

{
  "transform_id": "my_transform",
  "source": {
    "index": "logs-*"
  },
  "dest": {
    "index": "transformed-logs"
  },
  "docs": [
    {
      "message": "User login successful",
      "timestamp": "2026-01-15T10:00:00Z",
      "user_id": "user123"
    }
  ]
}
Field Description Notes
transform_id Identifier for the transform Must be unique within the cluster
source.index Source index pattern to transform Supports wildcards like logs-*
dest.index Destination index for transformed data Must not exist before running the transform
docs Sample documents to preview Used for testing and validation
Note: The docs field is used for previewing results and does not execute the full transform.
Warning: Previewing does not persist data to the destination index.

Previewing Transform Results - Key Points

  • Use the docs parameter to simulate transform execution
  • Preview does not write to the destination index
  • Validates transform configuration before full execution
  • Helps identify potential issues in data mapping or aggregation logic

9. Monitoring Transform Progress

Example: Check Transform Status

GET /_transform/<transform_id>/_stats
Field Description Values
transform_id Identifier of the transform my_transform
state Current state of the transform started, stopped, failed
checkpoint Last processed checkpoint 1634567890
Note: The state field indicates whether the transform is currently running or stopped.

Example: Get Transform Stats

{
  "id": "my_transform",
  "state": "started",
  "stats": {
    "documents_processed": 12500,
    "documents_failed": 0,
    "last_checkpoint": "1634567890"
  }
}

10. Handling Transform Errors

Error Type Description Handling Strategy Recovery Action
MappingException Occurs when a field mapping is invalid or missing in the transform Validate field mappings before applying transforms Update mapping configuration or add missing fields
ScriptException Thrown when a script used in the transform fails to execute Test scripts in isolation before integrating into transforms Fix or replace the faulty script logic
PipelineException Arises when a transform pipeline fails to process data correctly Use logging and monitoring to identify pipeline stages causing issues Adjust or remove problematic stages in the pipeline
TimeoutException Occurs when a transform operation exceeds the configured timeout limit Optimize transform logic or increase timeout settings Implement retry logic with exponential backoff

Example: Error Handling in Transform Script

try {
  const transformedData = transform(input);
} catch (error) {
  if (error.name === 'MappingException') {
    logger.error('Field mapping error:', error.message);
    // Log and skip invalid records
  } else if (error.name === 'ScriptException') {
    logger.error('Script execution error:', error.message);
    // Retry or alert on script failure
  }
}
Note: Use structured logging to capture transform errors for debugging and monitoring.
Warning: Unhandled transform errors can cause data loss or pipeline failures. Always implement error handling.

11. Using Transform Checkpointing

Transform checkpointing in the ELK Stack allows you to resume data transforms from a previous state, ensuring that processing can continue after interruptions or system failures.

Property Description Default Value Notes
checkpoint Indicates whether to enable checkpointing for the transform false Set to true to enable checkpointing
checkpoint_interval Interval at which to create checkpoints (in seconds) 30 Adjust based on data volume and processing frequency
max_checkpoint_interval Maximum time interval between checkpoints (in seconds) 60 Ensures checkpoints are not too infrequent

Example: Enabling Transform Checkpointing

{
  "id": "my_transform",
  "source": {
    "index": ["logs-*"]
  },
  "dest": {
    "index": "transformed-logs"
  },
  "checkpoint": {
    "enabled": true,
    "interval": 30
  }
}
Note: Checkpointing is particularly useful for long-running transforms where data ingestion may be intermittent or where system restarts are expected.

12. Optimizing Transform Performance

Transform Type Performance Considerations Optimization Tips
Ingest Pipeline Processing overhead on each document Minimize the number of processors in a pipeline. Use if conditions to avoid unnecessary processing.
Scripted Transforms JavaScript engine startup cost per transform Cache results where possible. Avoid complex logic in scripts.
Painless Scripting Execution time increases with complexity Use built-in functions instead of custom logic. Avoid nested loops.

Example: Ingest Pipeline Optimization

{
  "pipeline": {
    "id": "optimized-pipeline",
    "description": "Optimized pipeline for log processing",
    "processors": [
      {
        "if": "ctx.log_level == 'ERROR'",
        "set": {
          "error_count": 1
        }
      },
      {
        "remove": {
          "field": "temp_field"
        }
      }
    ]
  }
}
Note: Use if conditions to filter out unnecessary processing for better performance.

Key Performance Metrics

  • Transform latency (time to process a document)
  • Throughput (documents per second)
  • Resource utilization (CPU, memory)

Was this article helpful?

Related Cheat Sheets