教程：使用自定义摄取管道转换数据
编辑教程：使用自定义摄取管道转换数据编辑
本教程介绍如何将自定义摄取管道添加到 Elastic 集成。自定义管道可用于添加自定义数据处理，例如添加字段、模糊化敏感信息等等。
场景： 您有 Elastic Agent 使用系统集成收集系统指标。
目标： 添加一个自定义摄取管道，在索引 Elasticsearch 文档之前向每个文档添加一个新字段。
步骤 1：创建自定义摄取管道编辑
创建一个自定义摄取管道，该管道将由默认集成管道调用。在本教程中，我们将创建一个管道，该管道将向我们的文档添加一个新字段。
- 在 Kibana 中，导航至 堆栈管理 → 摄取管道 → 创建管道 → 新管道。
- 为您的管道命名。我们将此管道命名为
add_field。 -
选择 添加处理器。填写以下信息
- 处理器："设置"
- 字段：
test -
值：
true设置处理器 设置文档字段并将其与指定的值关联。
- 单击 添加。
- 单击 创建管道。
步骤 2：应用您的摄取管道编辑
通过从默认摄取管道调用自定义管道，将自定义管道添加到集成。自定义管道将在默认管道之后运行，但在最终管道之前运行。
编辑集成编辑
从 编辑集成 工作流程中将自定义管道添加到集成。在添加自定义管道之前，必须先配置并安装集成。要进入此工作流程，请执行以下操作
- 导航至 Fleet
- 选择相关的 Elastic Agent 策略
- 搜索您要编辑的集成
- 选择 操作 → 编辑集成
选择数据流编辑
大多数集成写入多个数据流。您需要分别将自定义管道添加到每个数据流。
- 找到您要编辑的第一个数据流，然后选择 更改默认值。在本教程中，请找到名为 从系统实例收集指标 的数据流配置。
-
滚动至 系统 CPU 指标，并在 高级选项 下选择 添加自定义管道。
这将带您进入 堆栈管理 中的 创建管道 工作流程。
添加管道编辑
添加您在步骤一中创建的管道。
-
选择 添加处理器。填写以下信息
- 处理器："管道"
- 管道名称："add_field"
- 值：
true
- 单击 创建管道 以返回 编辑集成 页面。
滚动数据流（可选）编辑
要使管道更改立即生效，您必须滚动数据流。如果您不这样做，更改将不会在下次计划的滚动之前生效。选择 立即应用并滚动。
数据流滚动后，请注意自定义摄取管道的名称。在本教程中，它是 metrics-system.cpu@custom。名称遵循模式 <type>-<dataset>@custom
- 类型：
metrics - 数据集：
system.cpu - 自定义摄取管道指定：
@custom
重复编辑
将自定义摄取管道添加到您要更新的任何其他数据流。
步骤 3：测试摄取管道（可选）编辑
在测试管道之前，请留出一些时间让新数据被摄取。在一个新窗口中，打开 Kibana 并导航至 Kibana 开发工具。
使用 存在查询 确保新字段 "test" 正在应用于文档。
|
要搜索的数据流。在本教程中，我们编辑了 |
|
|
在步骤一中设置的字段的名称。 |
如果您的自定义管道正常工作，此查询将返回至少一个文档。
步骤 4：添加自定义映射编辑
现在，您的 Elasticsearch 文档中正在设置一个新字段，您需要为该字段分配一个新的映射。使用 @custom 组件模板将自定义映射应用于集成数据流。
在 编辑集成 工作流程中，请执行以下操作
- 在 高级选项 下，选择铅笔图标以编辑
@custom组件模板。 -
为您的索引文档定义新字段。选择 添加字段 并添加以下信息
- 字段名称：
test - 字段类型：
Boolean
- 字段名称：
- 单击 添加字段。
- 单击 查看 以快速前进到查看步骤，然后单击 保存组件模板 以返回 编辑集成 工作流程。
- 要使更改立即生效，请选择 立即应用并滚动。
步骤 5：测试自定义映射（可选）编辑
在测试映射之前，请留出一些时间让新数据被摄取。在一个新窗口中，打开 Kibana 并导航至 Kibana 开发工具。
使用 获取字段映射 API 确保已应用自定义映射。
|
要搜索的数据流。在本教程中，我们编辑了 |
结果应包括指定字段的 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-fieldPUT _ingest/pipeline/logs@custom { "processors": [ { "set": { "description": "Custom field for all log events", "field": "my-logs-field", "value": "true" } } ] } -
留出一些时间让新数据被摄取，然后使用新的 存在查询 确认新字段 "my-logs-field" 正在应用于日志事件文档。
对于此示例，我们将检查系统集成
system.syslog数据集GET /logs-system.syslog-default/_search?pretty { "query": { "exists": { "field": "my-logs-field" } } }
应用了新管道后，此查询应返回至少一个文档。
您可以根据需要修改管道 API 请求，以在不同级别应用自定义处理。请参阅 摄取管道 了解更多信息。