如何在Logstash中读取CSV文件的每隔一行内容？

核心思路 #

Logstash 没有直接提供“隔行读取”的内置配置，但可以通过Ruby过滤器维护行计数器实现行筛选，再配合CSV过滤器解析制表符分隔的内容。

完整配置示例 #

input {
  file {
    path => "/path/to/your/tab-separated-file.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null" # 测试场景使用，避免记录读取位置导致重复测试问题
  }
}

filter {
  # 1. 初始化行计数器，筛选隔行内容
  ruby {
    code => '
      # 首次处理时初始化计数器为0
      @count ||= 0
      # 保留奇数行（第2、4、6...行，从0开始计数），若要保留偶数行则改为 @count % 2 == 0
      if @count % 2 == 0
        event.cancel # 取消当前行，不进入后续处理流程
      end
      @count += 1 # 计数器自增
    '
  }

  # 2. 解析制表符分隔的CSV内容
  csv {
    separator => "\t" # 指定制表符为字段分隔符
    columns => ["col1", "col2", "col3"] # 替换为你的实际列名
    skip_header => false # 若文件含表头且需保留，设为false；若要跳过表头则设为true
  }

  # 可选：移除计数器字段（无需保留时添加）
  mutate {
    remove_field => ["@count"]
  }
}

output {
  stdout {
    codec => rubydebug # 测试时输出到控制台，实际可替换为Elasticsearch等输出目标
  }
}

关键细节说明 #

• 行计数器逻辑：Ruby过滤器中的@count为每行事件的计数器，首次处理自动初始化。如需调整保留的行规则，修改取余判断条件即可。
• 表头处理：若CSV含表头且需保留表头、从第二行开始隔行读取，可调整Ruby代码：

  code => '
    @count ||= 0
    # 跳过表头行（第1行），从第2行开始保留奇数行
    if @count == 0
      @count +=1
    elsif @count % 2 != 0
      event.cancel
    end
    @count +=1
  '
  

  同时将CSV过滤器的skip_header设为false，避免重复跳过表头。
• CSV解析配置：columns需与CSV列一一对应；若不确定列数，可省略该配置，Logstash会自动生成column1、column2这类默认字段名。

替代方案（无需Ruby的简化版） #

如果不想使用Ruby过滤器，也可以通过环境变量配合简单判断实现，但灵活性稍差：

filter {
  mutate {
    add_field => {"count" => "${LS_COUNTER}"}
  }
  ruby {
    code => '
      count = event.get("count").to_i
      if count % 2 == 0
        event.cancel
      end
    '
  }
  # 后续CSV解析配置同上
}

启动Logstash时需手动初始化计数器：LS_COUNTER=0 bin/logstash -f your-config.conf，但每次重启都需要重置计数器，适合临时测试场景。

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