如何在Logstash的output模块使用mutate插件输出到两个Elasticsearch索引

问题根因 #

Logstash的input、filter、output三个阶段只能运行对应类型的插件，mutate属于filter类插件，写在output段不会被执行，这是配置不生效的根本原因。

推荐实现方案 #

方案1：单pipeline使用clone过滤器（最常用，适配绝大多数场景） #

在原有filter解析逻辑结束后，用clone插件复制一份完全相同的事件，给克隆事件打上专属标识，分别处理两类事件的字段后，在output阶段按标识路由到不同ES索引即可，配置示例如下：

filter {
  # 此处保留你原有日志解析的全部配置
  # ...原有解析逻辑...

  # 解析完成后克隆事件
  clone {
    add_tag => ["for_second_index"] # 给克隆出的事件打标签做区分
  }

  # 仅对要写入第二个索引的克隆事件做字段调整
  if "for_second_index" in [tags] {
    mutate {
      remove_field => ["待删除字段1", "待删除字段2"]
      add_field => {
        "新增字段1" => "自定义值1"
        "新增字段2" => "自定义值2"
      }
    }
  }
}

output {
  # 原始全量字段事件写入第一个索引
  if "for_second_index" not in [tags] {
    elasticsearch {
      hosts => ["http://你的ES地址:9200"]
      index => "第一个索引名称"
      # 其他原有ES配置如账号密码、证书等保留
    }
  }
  # 处理后的事件写入第二个索引
  if "for_second_index" in [tags] {
    elasticsearch {
      hosts => ["http://你的ES地址:9200"]
      index => "第二个索引名称"
      # 按需配置其他ES参数
    }
  }
}

方案2：多pipeline级联（适合复杂日志处理场景） #

如果你的日志处理逻辑链路很长，建议拆分多pipeline解耦维护，两个pipeline分别负责写入两个索引，配置如下：

1. 首先修改pipelines.yml定义pipeline关联关系：

- pipeline.id: process_and_write_first_index
  path.config: "/你的配置路径/pipeline1.conf"
- pipeline.id: modify_and_write_second_index
  path.config: "/你的配置路径/pipeline2.conf"

2. pipeline1.conf（原解析逻辑+写第一个索引+转发事件）：

input {
  # 原有输入配置
}
filter {
  # 原有日志解析配置
}
output {
  elasticsearch {
    # 第一个索引的全量写入配置
  }
  # 转发原始事件给第二个pipeline处理
  pipeline {
    send_to => "modify_and_write_second_index"
  }
}

3. pipeline2.conf（字段调整+写第二个索引）：

input {
  pipeline {
    address => "modify_and_write_second_index"
  }
}
filter {
  mutate {
    remove_field => ["待删除字段"]
    add_field => {"新增字段" => "自定义值"}
  }
}
output {
  elasticsearch {
    # 第二个索引的写入配置
  }
}

优化提示 #

如果需要删除的字段数量远大于保留的字段，可替换mutate remove_field为prune过滤器的whitelist_names参数，直接指定要保留的字段列表，处理效率更高。

内容的提问来源于stack exchange，提问作者Orkun