---
title: Tutorial: Transform data with custom ingest pipelines
description: This tutorial explains how to add a custom ingest pipeline to an Elastic Integration. Custom pipelines can be used to add custom data processing, like...
url: https://www.elastic.co/elastic/docs-builder/docs/3028/reference/fleet/data-streams-pipeline-tutorial
products:
  - Elastic Agent
  - Fleet
applies_to:
  - Elastic Cloud Serverless: Generally available
  - Elastic Stack: Generally available
---

# Tutorial: Transform data with custom ingest pipelines
This tutorial explains how to add a custom ingest pipeline to an Elastic Integration. Custom pipelines can be used to add custom data processing, like adding fields, obfuscate sensitive information, and more.
**Scenario:** You have Elastic Agents collecting system metrics with the System integration.
**Goal:** Add a custom ingest pipeline that adds a new field to each Elasticsearch document before it is indexed.

## Step 1: Create a custom ingest pipeline

Create a custom ingest pipeline that will be called by the default integration pipeline. In this tutorial, we’ll create a pipeline that adds a new field to our documents.
1. In Kibana, go to the **Ingest Pipelines** management page using the navigation menu or the [global search field](https://www.elastic.co/elastic/docs-builder/docs/3028/explore-analyze/find-and-organize/find-apps-and-objects).
2. **Create pipeline** → **New pipeline**.
3. Name your pipeline. We’ll call this one, `add_field`.
4. Select **Add a processor**. Fill out the following information:
   - Processor: "Set"
- Field: `test`
- Value: `true`
  The [Set processor](https://docs-v3-preview.elastic.dev/elastic/docs-builder/docs/3028/reference/enrich-processor/set-processor) sets a document field and associates it with the specified value.
5. Click **Add**.
6. Click **Create pipeline**.


## Step 2: Apply your ingest pipeline

Add a custom pipeline to an integration by calling it from the default ingest pipeline. The custom pipeline will run after the default pipeline but before the final pipeline.

### Edit integration

Add a custom pipeline to an integration from the **Edit integration** workflow. The integration must already be configured and installed before a custom pipeline can be added. To enter this workflow, do the following:
1. Navigate to **Fleet**
2. Select the relevant Elastic Agent policy
3. Search for the integration you want to edit
4. Select **Actions** → **Edit integration**


### Select a data stream

Most integrations write to multiple data streams. You’ll need to add the custom pipeline to each data stream individually.
1. Find the first data stream you wish to edit and select **Change defaults**. For this tutorial, find the data stream configuration titled, **Collect metrics from System instances**.
2. Scroll to **System CPU metrics** and under **Advanced options** select **Add custom pipeline**.
   This will take you to the **Create pipeline** workflow.


### Add the pipeline

Add the pipeline you created in step one.
1. Select **Add a processor**. Fill out the following information:
   - Processor: "Pipeline"
- Pipeline name: "add_field"
- Value: `true`
2. Click **Create pipeline** to return to the **Edit integration** page.

After saving, note the name of the custom ingest pipeline. In this tutorial, it’s `metrics-system.cpu@custom`. The name follows the pattern `<type>-<dataset>@custom`:
- type: `metrics`
- dataset: `system.cpu`
- Custom ingest pipeline designation: `@custom`


### Repeat

Add the custom ingest pipeline to any other data streams you wish to update.

## Step 3: Test the ingest pipeline (optional)

Allow time for new data to be ingested before testing your pipeline. In a new window, open Kibana and navigate to **Kibana Dev tools**.
Use an [exists query](https://docs-v3-preview.elastic.dev/elastic/docs-builder/docs/3028/reference/query-languages/query-dsl/query-dsl-exists-query) to ensure that the new field, "test" is being applied to documents.
```json

{
  "query": {
    "exists": {
      "field": "test" <2>
    }
  }
}
```

If your custom pipeline is working correctly, this query will return at least one document.

## Step 4: Add custom mappings

Now that a new field is being set in your Elasticsearch documents, you’ll want to assign a new mapping for that field. Use the `@custom` component template to apply custom mappings to an integration data stream.
<note>
  Mapping and template changes are applied when a new backing index is created. To apply updated mappings immediately, roll over the data stream, then continue.For more information, refer to [Index basics](https://www.elastic.co/elastic/docs-builder/docs/3028/manage-data/data-store/index-basics).
</note>

In the **Edit integration** workflow, do the following:
1. Under **Advanced options** select the pencil icon to edit the `@custom` component template.
2. Define the new field for your indexed documents. Select **Add field** and add the following information:
   - Field name: `test`
- Field type: `Boolean`
3. Click **Add field**.
4. Click **Review** to fast-forward to the review step and click **Save component template** to return to the **Edit integration** workflow.
5. To apply the new mapping immediately, roll over the data stream. You can do this with the rollover API, for example:
   ```json
   ```


## Step 5: Test the custom mappings (optional)

Allow time for new data to be ingested before testing your mappings. In a new window, open Kibana and navigate to **Kibana Dev tools**.
Use the [Get field mapping API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-get-mapping) to ensure that the custom mapping has been applied.
```json
```

The result should include `type: "boolean"` for the specified field.
```json
".ds-metrics-system.cpu-default-2022.08.10-000002": {
  "mappings": {
    "test": {
      "full_name": "test",
      "mapping": {
        "test": {
          "type": "boolean"
        }
      }
    }
  }
}
```


## Step 6: Add an ingest pipeline for a data type

The previous steps demonstrated how to create a custom ingest pipeline that adds a new field to each Elasticsearch document generated for the Systems integration CPU metrics (`system.cpu`) dataset.
You can create an ingest pipeline to process data at various levels of customization. An ingest pipeline processor can be applied:
- Globally to all events
- To all events of a certain type (for example `logs` or `metrics`)
- To all events of a certain type in an integration
- To all events in a specific dataset

Let’s create a new custom ingest pipeline `logs@custom` that processes all log events.
1. Open Kibana and navigate to **Kibana Dev tools**.
2. Run a [pipeline API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-ingest-put-pipeline) request to add a new field `my-logs-field`:
   ```json

   {
     "processors": [
       {
         "set": {
           "description": "Custom field for all log events",
           "field": "my-logs-field",
           "value": "true"
         }
       }
     ]
   }
   ```
3. Allow some time for new data to be ingested, and then use a new [exists query](https://docs-v3-preview.elastic.dev/elastic/docs-builder/docs/3028/reference/query-languages/query-dsl/query-dsl-exists-query) to confirm that the new field "my-logs-field" is being applied to log event documents.
   For this example, we’ll check the System integration `system.syslog` dataset:
   ```json

   {
     "query": {
       "exists": {
         "field": "my-logs-field"
       }
     }
   }
   ```

With the new pipeline applied, this query should return at least one document.
You can modify your pipeline API request as needed to apply custom processing at various levels. Refer to [Ingest pipelines](/elastic/docs-builder/docs/3028/reference/fleet/data-streams#data-streams-pipelines) to learn more.