如何验证Logstash索引Elasticsearch时是否跳过空行？

1. 检查Input插件配置 #

先确认Logstash的file input配置里是否设置了ignore_empty_lines => true，这个参数会直接跳过空行。如果有该配置，改成ignore_empty_lines => false后重新运行，再对比Elasticsearch的文档数量。

2. 输出到本地文件对比行数 #

修改Logstash配置，在output部分新增一个file输出，将所有处理后的事件原样导出到本地文件：

output {
  elasticsearch {
    # 保留你的原有ES配置
  }
  file {
    path => "/tmp/logstash-processed.log"
    codec => line { format => "%{message}" }
  }
}

完成导入后，分别统计原日志文件和处理后文件的行数：

# 统计原文件行数
wc -l your-original-log.log
# 统计处理后文件行数
wc -l /tmp/logstash-processed.log

如果处理后的文件行数比原文件少1，且原文件确实存在1个空行，即可确认空行被忽略。

3. 标记空行并在ES中查询 #

在filter阶段添加逻辑，给空行事件添加专属标记字段，之后在ES中查询该标记的文档数量：

filter {
  # 匹配空行（包括仅含空白字符的行）
  if [message] =~ /^\s*$/ {
    mutate {
      add_field => { "is_empty_line" => "true" }
    }
  }
}

导入完成后，在ES中执行统计查询：

GET /your-index/_count
{
  "query": {
    "term": {
      "is_empty_line": "true"
    }
  }
}

如果统计结果为0，而原文件存在1个空行，说明空行未被导入；如果结果为1，则需要排查其他导致文档缺失的原因（比如某行数据格式错误被过滤）。

4. 实时查看stdout输出 #

临时修改Logstash配置，将输出改为stdout，直接观察处理的每一行内容：

output {
  stdout {
    codec => line { format => "%{message}" }
  }
}

运行Logstash并将输出保存到文件：

bin/logstash -f your-config.conf > /tmp/logstash-stdout.log

统计该文件行数并与原文件对比，同时可以直接打开文件查看是否缺失了空行。

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