Logstash导入CSV忽略换行，仅生成单条ElasticSearch记录问题排查

我尝试通过Logstash将CSV文件上传至ElasticSearch，但发现Logstash忽略了CSV文件中的换行符，最终仅在ElasticSearch中生成1条记录。以下是相关文件内容与配置：

CSV文件内容 #

root@debian:~# cat jmeter-localhost-dummy.csv
23c2a43199061f723b85832e49d5ff1cdc6590b085b1fdf17629e36d70c1ff68@@2023-11-08 11:10:09,689@@influxdb@@test@@dummy@@localhost@@INFO@@o.a.j.s.SampleResult: sampleresult.nanoThreadSleep=5000
cb17fb2a09a3f3cdb756792b871b488c551172cafe3bcf14aa9c1ed777a34df6@@2023-11-08 11:10:09,734@@influxdb@@test@@dummy@@localhost@@INFO@@o.a.j.t.ThreadGroup: Started thread group number 1
...
And so on ..

Logstash配置文件内容 #

root@debian:~# cat /etc/logstash/conf.d/http-pipeline.conf
input {
  http {
    host => "0.0.0.0"
    port => "3101"
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost/"]
    ssl => false
    ssl_certificate_verification => false
    index => "logstash"
    document_id => "%{sha256}"
  }
  stdout {
  }
}

filter {
  csv {
    skip_header => "true"
    columns => ["sha256", "datetime", "level", "campagne", "env", "tir", "injecteur", "message"]
    separator => "@@"
  }
  date {
    match => [ "datetime", "yyyy-MM-dd HH:mm:ss','SSS" ]
    timezone => "Europe/Paris"
    target => "@timestamp"
  }
  mutate {
    remove_field => [ "datetime" ]
  }
}

问题原因 #

通过Logstash的http输入接收CSV文件时，整个文件会被作为单条事件传入，默认情况下Logstash不会自动按换行符拆分事件，导致CSV过滤器仅处理第一行内容，后续行被忽略。

解决步骤 #

1. 添加split过滤器拆分事件 #

在csv过滤器之前，添加split过滤器，按换行符拆分整个请求内容为多条独立事件：

filter {
  # 先按换行符拆分事件
  split {
    field => "message"
    terminator => "\n"
  }
  
  # 原有CSV解析逻辑保持不变
  csv {
    skip_header => "true"
    columns => ["sha256", "datetime", "level", "campagne", "env", "tir", "injecteur", "message"]
    separator => "@@"
  }
  
  date {
    match => [ "datetime", "yyyy-MM-dd HH:mm:ss','SSS" ]
    timezone => "Europe/Paris"
    target => "@timestamp"
  }
  
  mutate {
    remove_field => [ "datetime" ]
  }
}

• field => "message"：指定拆分的目标字段（http输入默认将请求体存入message字段）
• terminator => "\n"：按换行符拆分每条CSV记录

2. 过滤空行（可选） #

如果CSV文件末尾存在空行，拆分后会生成无效空事件，可添加条件过滤：

filter {
  split {
    field => "message"
    terminator => "\n"
  }
  
  # 过滤空行或仅含空白字符的行
  if [message] =~ /^\s*$/ {
    drop {}
  }
  
  # 后续CSV、date、mutate过滤器保持不变
}

3. 验证效果 #

修改配置后重启Logstash，重新上传CSV文件，此时每条行会被拆分为独立事件，经过解析后生成多条ElasticSearch记录，同时stdout会打印每条事件，方便实时排查。

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