Logstash导入CSV时替换列名空格及解决数据未索引问题

尝试将带有空格表头的CSV文件导入Elasticsearch现有索引，需要在Logstash导入过程中将带空格的列名（例如"MOBILE NO"）重命名为下划线形式（如"MOBILE_NO"）。配置Pipeline后，日志显示管道正常运行，但没有数据被索引到Elasticsearch，也无后续输出内容。

当前Windows环境下的Pipeline配置如下：

input{
file{
    path => "E:\abc\xyz\johny.*csv"
    start_position => "beginning"
    sincedb_path => "NUL"
}
}
filter{
csv{
    separator => ","
    columns => ["STATUS_TYPE","GENDER","AGE","MOBILE NO","HOUSE NO"]
}    
ruby  {
    code => "
            hash = event.to_hash
            hash.each{|k,v| event.set(k.gsub(/ /,'_'),v) && event.remove(k) if k.include?(' ')  }
        "
}
}
output{
elasticsearch{
    hosts => "http://localhost:9200"
    index => "heaven_search"
}
stdout{
    codec => rubydebug
}
}

1. 修复Windows文件路径问题 #

Windows路径中的反斜杠需要转义，否则Logstash会将其识别为转义字符，导致无法找到目标文件。可以用两种方式修正：

• 使用双反斜杠：path => "E:\\abc\\xyz\\johny.*csv"
• 改用正斜杠：path => "E:/abc/xyz/johny.*csv"

2. 确保CSV解析逻辑正确 #

如果你的CSV文件第一行是表头，需要在csv filter中添加skip_header => true，避免把表头当成数据行处理。同时要保证columns定义的列名顺序和CSV数据列完全对应：

csv{
    separator => ","
    skip_header => true
    columns => ["STATUS_TYPE","GENDER","AGE","MOBILE NO","HOUSE NO"]
}

如果不想手动指定列名，也可以开启自动识别表头：

csv{
    separator => ","
    autodetect_column_names => true
}

3. 修正Ruby重命名逻辑 #

原Ruby代码在遍历Event哈希时直接修改原哈希，可能导致遍历异常。改为遍历键的副本，避免结构变动影响遍历：

ruby {
    code => "
        event.to_hash.keys.each do |k|
            if k.include?(' ')
                new_key = k.gsub(/\s+/, '_')
                event.set(new_key, event.get(k))
                event.remove(k)
            end
        end
    "
}

4. 补充调试与验证 #

• 启动Logstash时添加--log.level debug参数，查看详细日志，确认文件是否被读取、CSV解析是否有错误。
• 确保CSV文件没有被其他程序（如Excel）锁定，否则Logstash无法读取内容。
• 访问http://localhost:9200确认Elasticsearch服务正常运行。

修正后的完整配置 #

input{
    file{
        path => "E:/abc/xyz/johny.*csv"
        start_position => "beginning"
        sincedb_path => "NUL"
        stat_interval => 10 # 每10秒检查一次文件变化
    }
}
filter{
    csv{
        separator => ","
        skip_header => true
        columns => ["STATUS_TYPE","GENDER","AGE","MOBILE NO","HOUSE NO"]
    }    
    ruby  {
        code => "
            event.to_hash.keys.each do |k|
                if k.include?(' ')
                    new_key = k.gsub(/\s+/, '_')
                    event.set(new_key, event.get(k))
                    event.remove(k)
                end
            end
        "
    }
}
output{
    elasticsearch{
        hosts => "http://localhost:9200"
        index => "heaven_search"
        manage_template => true # 自动创建索引模板（可选）
    }
    stdout{
        codec => rubydebug
    }
}

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