如何在Logstash中将被丢弃的CSV行输出到文件

数据老陈

2026-7-25

记录Logstash中被丢弃的不符合邮箱格式的行

要把不符合邮箱正则的行记录到文件，有两种实用方案，可根据你的现有配置选择：

方案1：通过Output分流推荐

直接去掉Filter里的drop {}，在Output块中做条件判断，将符合格式的行导入ES，不符合的写入文件。这种方式更贴合Logstash的数据流设计，性能更稳定：

output {
  # 符合邮箱格式的行输出到Elasticsearch
  if [email] =~ /^[_a-z0-9-]+(\.[_a-z0-9-]+)*@[a-z0-9-]+(\.[a-z0-9-]+)*(\.[a-z]{2,3})$/ {
    elasticsearch {
      hosts => ["你的ES地址:9200"]
      index => "目标索引名"
      # 补充其他ES配置，比如认证参数等
    }
  }
  # 不符合格式的行写入日志文件
  else {
    file {
      path => "/你要保存的路径/invalid_emails.log"
      # 自定义输出格式，比如包含时间、错误邮箱和原始行内容
      codec => line { format => "%{@timestamp} | 无效邮箱: %{email} | 原始数据: %{message}" }
    }
  }
}

方案2：保留原有Filter的Drop逻辑

如果必须保留Filter里的drop {}操作，可以在执行drop前用Ruby插件把行数据写入文件：

filter {
  if [email] !~ /^[_a-z0-9-]+(\.[_a-z0-9-]+)*@[a-z0-9-]+(\.[a-z0-9-]+)*(\.[a-z]{2,3})$/ {
    ruby {
      code => '
        # 打开文件追加写入，确保路径存在且Logstash有读写权限
        file = File.open("/你要保存的路径/invalid_emails.log", "a")
        file.write(event.to_s + "\n")
        file.close
      '
    }
    drop {}
  }
}