« 教程：自定义数据保留策略为 Fleet 管理的数据流启用和禁用高级索引功能 »

› › ›

教程：使用自定义摄取管道转换数据

编辑

教程：使用自定义摄取管道转换数据

编辑

本教程解释了如何向 Elastic 集成添加自定义摄取管道。自定义管道可用于添加自定义数据处理，例如添加字段、混淆敏感信息等。

场景： 您已使用系统集成通过 Elastic Agent 收集系统指标。

目标： 添加自定义摄取管道，在每个 Elasticsearch 文档被索引之前向其添加新字段。

步骤 1：创建自定义摄取管道

编辑

创建一个自定义摄取管道，该管道将由默认集成管道调用。在本教程中，我们将创建一个向文档添加新字段的管道。

在 Kibana 中，导航至 堆栈管理 → 摄取管道 → 创建管道 → 新管道。
命名您的管道。我们将其命名为 add_field。
选择 添加处理器。填写以下信息
- 处理器：“Set”
- 字段：test
- 值：true
  
  Set 处理器设置一个文档字段并将其与指定的值相关联。
单击添加。
单击 创建管道。

步骤 2：应用您的摄取管道

编辑

通过从默认摄取管道调用自定义管道，将自定义管道添加到集成。自定义管道将在默认管道之后但在最终管道之前运行。

编辑集成

编辑

从 编辑集成 工作流程向集成添加自定义管道。必须先配置并安装集成，然后才能添加自定义管道。要进入此工作流程，请执行以下操作

导航至 Fleet
选择相关的 Elastic Agent 策略
搜索您要编辑的集成
选择操作 → 编辑集成

选择数据流

编辑

大多数集成会写入多个数据流。您需要单独向每个数据流添加自定义管道。

找到您要编辑的第一个数据流，然后选择 更改默认值。在本教程中，找到标题为 从系统实例收集指标 的数据流配置。
滚动到 系统 CPU 指标，然后在 高级选项 下，选择 添加自定义管道。

这将带您进入 堆栈管理 中的 创建管道 工作流程。

添加管道

编辑

添加您在第一步中创建的管道。

选择 添加处理器。填写以下信息
- 处理器：“Pipeline”
- 管道名称：“add_field”
- 值：true
单击 创建管道 返回到 编辑集成 页面。

滚动数据流（可选）

编辑

要使管道更改立即生效，您必须滚动数据流。如果您不这样做，则更改将在下一次计划滚动之前不会生效。选择 立即应用并滚动。

在数据流滚动后，记下自定义摄取管道的名称。在本教程中，它是 metrics-system.cpu@custom。该名称遵循模式 <type>-<dataset>@custom

类型：metrics
数据集：system.cpu
自定义摄取管道指定：@custom

重复

编辑

将自定义摄取管道添加到您希望更新的任何其他数据流。

步骤 3：测试摄取管道（可选）

编辑

在测试您的管道之前，请等待一段时间以摄取新数据。在新窗口中，打开 Kibana 并导航至 Kibana 开发工具。

使用 exists 查询以确保新字段 "test" 正在应用于文档。

GET metrics-system.cpu-default/_search 
{
  "query": {
    "exists": {
      "field": "test" 
    }
  }
}

	要搜索的数据流。在本教程中，我们编辑了 `metrics-system.cpu` 类型和数据集。`default` 是默认命名空间。将这三个组合在一起，得到的数据流名称为 `metrics-system.cpu-default`。
	在步骤 1 中设置的字段的名称。

如果您的自定义管道正常工作，此查询将返回至少一个文档。

步骤 4：添加自定义映射

编辑

现在，在您的 Elasticsearch 文档中设置了一个新字段，您需要为该字段分配一个新的映射。使用 @custom 组件模板将自定义映射应用于集成数据流。

在 编辑集成 工作流程中，执行以下操作

在 高级选项 下，选择铅笔图标以编辑 @custom 组件模板。
为您的索引文档定义新字段。选择 添加字段 并添加以下信息
- 字段名称：test
- 字段类型：Boolean
单击 添加字段。
单击审查以快进到审查步骤，然后单击 保存组件模板 返回到 编辑集成 工作流程。
要使更改立即生效，请选择 立即应用并滚动。

步骤 5：测试自定义映射（可选）

编辑

在测试您的映射之前，请等待一段时间以摄取新数据。在新窗口中，打开 Kibana 并导航至 Kibana 开发工具。

使用获取字段映射 API 以确保已应用自定义映射。

GET metrics-system.cpu-default/_mapping/field/test

要搜索的数据流。在本教程中，我们编辑了 metrics-system.cpu 类型和数据集。default 是默认命名空间。将这三个组合在一起，得到的数据流名称为 metrics-system.cpu-default。

结果应包括指定字段的 type: "boolean"。

".ds-metrics-system.cpu-default-2022.08.10-000002": {
  "mappings": {
    "test": {
      "full_name": "test",
      "mapping": {
        "test": {
          "type": "boolean"
        }
      }
    }
  }
}

步骤 6：为数据类型添加摄取管道

编辑

前面的步骤演示了如何创建一个自定义摄取管道，该管道将一个新字段添加到为系统集成 CPU 指标 (system.cpu) 数据集生成的每个 Elasticsearch 文档中。

您可以创建摄取管道以在各种自定义级别处理数据。可以应用摄取管道处理器

全局应用于所有事件
应用于某种类型的所有事件（例如 logs 或 metrics）
应用于集成中某种类型的所有事件
应用于特定数据集中的所有事件

让我们创建一个新的自定义摄取管道 logs@custom，该管道处理所有日志事件。

打开 Kibana 并导航至 Kibana 开发工具。

运行一个管道 API 请求以添加新字段 my-logs-field

PUT _ingest/pipeline/logs@custom
{
  "processors": [
    {
      "set": {
        "description": "Custom field for all log events",
        "field": "my-logs-field",
        "value": "true"
      }
    }
  ]
}

等待一段时间以摄取新数据，然后使用新的 exists 查询以确认新字段 "my-logs-field" 正在应用于日志事件文档。

在此示例中，我们将检查系统集成 system.syslog 数据集
```
GET /logs-system.syslog-default/_search?pretty
{
  "query": {
    "exists": {
      "field": "my-logs-field"
    }
  }
}
```

应用新管道后，此查询应返回至少一个文档。

您可以根据需要修改您的管道 API 请求，以在各个级别应用自定义处理。请参阅摄取管道以了解更多信息。

« 教程：自定义数据保留策略为 Fleet 管理的数据流启用和禁用高级索引功能 »