如何使用Logstash处理空格分隔文本文件？含报错与配置排查

首先，咱们先搞定你遇到的插件加载错误——你配置里用的txt filter插件根本不是Logstash自带的，这就是报错的核心原因！Logstash里处理分隔符文本的标准插件是csv，它支持自定义分隔符，完全能满足你处理空格分隔文件的需求。

接下来咱们一步步修正配置，完美解析你的日志文件：

1. 替换不存在的txt插件为csv插件 #

把filter块里的txt换成csv，并指定正确的分隔符为空格。另外，你配置里的columns字段和实际日志的表头完全不匹配，必须改成和日志第一行一致的字段名。

2. 配置正确的字段映射 #

你的日志表头是：
maker model mileage manufacture_year engine_displacement engine_power body_type color_slug stk_year transmission door_count seat_count fuel_type date_created date_last_seen price_eur
所以columns里要严格对应这些字段。

3. 优化配置：处理日期和数值类型 #

日志里的日期字段是带时区的格式，咱们可以用date插件解析成Elasticsearch能识别的日期类型；另外像里程、价格这些数值字段，也可以用mutate插件转成数值类型，方便后续分析；还有日志里的None值，咱们可以转成null，避免存成无意义的字符串。

完整的正确配置示例 #

input {
    file {
        path => "/home/elk/data/logklda.txt"
        start_position => "beginning" 
        sincedb_path => "/dev/null" # 每次启动都从头读取，适合测试场景
    }
} 

filter{
    # 用csv插件处理空格分隔的文本
    csv {
        separator => " "
        columns => ["maker", "model", "mileage", "manufacture_year", "engine_displacement", "engine_power", "body_type", "color_slug", "stk_year", "transmission", "door_count", "seat_count", "fuel_type", "date_created", "date_last_seen", "price_eur"]
        skip_header => true # 跳过第一行表头，避免被当成数据解析
    }

    # 解析date_created字段为标准时间格式
    date {
        match => ["date_created", "yyyy-MM-dd HH:mm:ss.SSSSSSZ"]
        target => "@timestamp" # 可选：将该字段设为事件的默认时间戳
    }
    # 解析date_last_seen字段
    date {
        match => ["date_last_seen", "yyyy-MM-dd HH:mm:ss.SSSSSZ"]
        target => "date_last_seen"
    }

    # 转换字段为对应数值类型，并处理None值
    mutate {
        convert => {
            "mileage" => "integer"
            "manufacture_year" => "integer"
            "engine_displacement" => "integer"
            "engine_power" => "integer"
            "door_count" => "integer"
            "seat_count" => "integer"
            "price_eur" => "float"
        }
        # 把字符串"None"替换为null
        gsub => [
            "body_type", "None", ""
        ]
        remove_field => ["message"] # 可选：移除原始的message字段，只保留解析后的结构化字段
    }
} 

output {
    elasticsearch {
        hosts => "localhost"
        index => "logklda"
        document_type => "car_data" # 建议改成更贴合数据的类型名
    }
    stdout { codec => rubydebug } # 用rubydebug格式输出，方便调试查看解析结果
}

关键注意事项 #

• 一定要加上skip_header => true，否则第一行表头会被当成一条数据解析
• 日期格式要和日志里的完全匹配，你的日志日期是2015-11-14 18:10:06.838319+00，对应的格式是yyyy-MM-dd HH:mm:ss.SSSSSSZ
• 测试时可以先注释掉elasticsearch输出，只保留stdout { codec => rubydebug }，确认解析结果正确后再写入Elasticsearch

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