如何使用Logstash实现数组中代码的CSV多列匹配并生成对应分类与描述字段？

你遇到的问题确实戳中了Logstash自带过滤器的局限性——csv过滤器主要用来解析事件内的CSV格式字段，translate过滤器只支持键值对的两列映射，都没法处理数组循环+多列关联的场景。这里推荐用Ruby过滤器来实现，它能完全自定义逻辑，完美满足你的需求。

核心思路 #

1. 预加载CSV到内存：在Logstash启动时，把CSV文件的内容加载成一个全局哈希表，键是indicator.code，值是包含category和description的嵌套哈希，这样后续处理事件时可以快速查找匹配。
2. 循环处理数组元素：对每个事件中的indicator.codes数组，遍历每个代码，去哈希表中匹配对应的分类和描述，最后把结果分别收集成数组，存入indicator.categories和indicator.descriptions字段。

具体配置示例 #

ruby {
  # 初始化块：Logstash启动时仅执行一次，加载CSV构建映射表
  init => '
    require "csv"
    # 替换成你的CSV文件绝对路径
    csv_path = "/opt/logstash/config/indicator_codes.csv"
    @code_mapping = {}
    # 读取CSV，分隔符用;（和你的CSV格式一致）
    CSV.foreach(csv_path, col_sep: ";") do |row|
      code = row[0]
      category = row[1]
      description = row[2]
      # 构建映射：code => { category: xxx, description: xxx }
      @code_mapping[code] = { category: category, description: description }
    end
  '
  # 处理每个事件的逻辑
  code => '
    # 获取事件中的indicator.codes数组
    codes = event.get("indicator.codes")
    # 确保是数组类型才处理（避免字段不存在或类型错误）
    if codes.is_a?(Array)
      categories = []
      descriptions = []
      # 遍历每个code
      codes.each do |code|
        # 查找映射表中的条目
        code_entry = @code_mapping[code]
        if code_entry
          # 收集对应的分类和描述
          categories << code_entry[:category]
          descriptions << code_entry[:description]
        else
          # 可选：如果code不存在，添加默认值（比如"Unknown"）
          # categories << "Unknown"
          # descriptions << "No description found for code #{code}"
        end
      end
      # 设置新字段，只有当数组非空时才设置（避免生成空字段）
      event.set("indicator.categories", categories) unless categories.empty?
      event.set("indicator.descriptions", descriptions) unless descriptions.empty?
    end
  '
}

关键细节说明 #

• 初始化块（init）：只在Logstash启动时执行一次，避免每次处理事件都重复读取CSV，大幅提升性能。要确保Logstash进程对CSV文件有读取权限。
• 数组处理逻辑：先判断indicator.codes是否为数组类型，防止字段不存在或格式错误导致的异常，然后遍历每个元素去映射表匹配。
• 无效code处理：如果数组中的某个code在CSV里找不到对应条目，默认会被忽略。如果需要标记这种情况，可以取消注释代码里的默认值部分。
• CSV格式兼容：代码里指定了分隔符为;，和你提供的CSV示例一致，如果你的CSV用其他分隔符，修改col_sep参数即可。

额外注意事项 #

• 如果CSV文件内容有更新，需要重启Logstash才能加载新的映射表（因为初始化块只在启动时执行）。如果需要动态更新映射，可以考虑结合定时任务或外部存储，但静态场景下重启完全足够。
• 确保indicator.codes字段确实是数组类型，如果你的日志里是字符串格式（比如"[3,120,148]"），需要先用mutate过滤器转换成数组：

  mutate {
    gsub => ["indicator.codes", "\[|\]", ""]
    split => ["indicator.codes", ","]
  }
  

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