---
title: 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/4384/reference/fleet/data-streams-pipeline-tutorial
products:
  - Elastic Agent
  - Elastic Documentation
  - Fleet
applies_to:
  - Elastic Cloud Serverless: Generally available
  - Elastic Stack: Generally available
---

# 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/4384/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/4384/reference/ingest-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.
<note>
  You can only add a custom pipeline to an integration that's already added to an Elastic Agent policy. The **Ingest pipelines** settings don't appear while you're adding an integration for the first time. If you haven't added the System integration yet, add it to an Elastic Agent policy and save it, then follow the steps in this section to edit it.
</note>


### Edit integration

Add a custom pipeline to an integration from the **Edit integration** workflow. To enter this workflow, do the following:
1. In Kibana, find **Fleet** in the navigation menu or use the [global search field](https://www.elastic.co/elastic/docs-builder/docs/4384/explore-analyze/find-and-organize/find-apps-and-objects), then select **Agent policies**.
2. Select the Elastic Agent policy that includes the integration you want to edit.
3. Find the integration in the list of integrations for the policy.
4. From the integration's `ellipsis` actions menu, select **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.
The custom pipeline is named after the data stream, so any change to it applies to every policy that uses this integration and writes to the same data stream.
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 expand **Advanced options**.
3. At the end of the advanced options, find the **Ingest pipelines** section. It lists the default ingest pipeline for the data stream, named `<type>-<dataset>-<integration version>`.
4. Below the default pipeline, select **Add custom pipeline**. This takes you to the **Create pipeline** workflow, with the pipeline name already set to `metrics-system.cpu@custom`.
   If the data stream already has a custom pipeline, **Add custom pipeline** doesn't appear. Instead, the custom pipeline is listed below the default pipeline. Select `pencil` **Edit pipeline** to add the processor from the next section to the existing custom pipeline.


### 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"
2. Click **Create pipeline** to return to the **Edit integration** page. If you're editing an existing custom pipeline, click **Save pipeline** instead.

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/4384/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/4384/manage-data/data-store/index-basics).
</note>

In the **Edit integration** workflow, return to **Collect metrics from System instances**, then expand **Advanced options** for **System cpu metrics**.
1. Find the **Mappings** section. This is where you add custom mappings for fields your ingest pipeline creates. It lists the component templates for the data stream.
2. Select **Add custom mappings**. This takes you to the component template workflow, with the template name already set to `metrics-system.cpu@custom`.
   If the data stream already has a custom component template, **Add custom mappings** doesn't appear. Select `pencil` **Edit mappings** on the `@custom` component template instead.
3. Define the new field for your indexed documents using the following information:
   - Field name: `test`
- Field type: `Boolean`
4. Click **Add field**.
5. Click **Review** to fast-forward to the review step and click **Create component template** to return to the **Edit integration** workflow. If you're editing an existing custom component template, click **Save component template** instead.
6. 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/4384/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/4384/reference/fleet/data-streams#data-streams-pipelines) to learn more.