如何配置Logstash将文档拆分为独立字段后写入Elasticsearch

你当前的Logstash配置没有配置解析规则，默认会将读取到的整行文本全部存入message字段，不会自动拆分JSON键值对。同时你导入的是Mongo导出的非标准JSON，包含NumberLong、ISODate等Mongo专属语法，需要先预处理为标准JSON才能正常解析。

input {
    file {
        path => "C:/Users/poler/Downloads/logstash-oss-7.10.0-windows-x86_64/files/file2.json"
        start_position => "beginning"
        sincedb_path => "NUL"
        codec => plain { charset => "UTF-8" }
    }
}
filter {
    # 预处理Mongo特殊语法，转为标准JSON格式
    mutate {
        gsub => [
            "message", "NumberLong\((\d+)\)", "\1",
            "message", "ISODate\(\"([^\"]+)\"\)", "\"\1\""
        ]
    }
    # 解析JSON内容，将所有键值对展开为独立字段
    json {
        source => "message"
        skip_on_invalid_json => true
    }
    # 转换时间字段为Logstash标准时间类型，方便ES自动识别为日期格式
    date {
        match => ["created", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"]
        target => "created"
        timezone => "UTC"
    }
    date {
        match => ["lastUpdated", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"]
        target => "lastUpdated"
        timezone => "UTC"
    }
    # 可选：删除不需要的原始message字段，减少冗余
    mutate {
        remove_field => ["message", "path", "host", "@version"]
    }
}
output {
    elasticsearch {
        action => "index"
        hosts => ["https://<hostIP>:9200"]
        user => "####"
        password => "####"
        ssl_certificate_verification => false
        cacert => "C:/Users/poler/Downloads/logstash-oss-7.10.0-windows-x86_64/elasticsearch-cert.pem"
        # 可选：使用Mongo中的_id作为ES文档的唯一ID，避免重复导入生成重复数据
        document_id => "%{_id}"
    }
    stdout{
        codec => rubydebug
    }
}

• 输入部分新增了UTF-8编码指定，避免中文或特殊字符乱码
• 新增filter逻辑：
  • 先通过正则替换把Mongo专属的NumberLong、ISODate语法转为标准JSON支持的格式，避免解析报错
  • 用json过滤器直接解析message字段的内容，自动将所有键值对展开为独立的顶级字段
  • 时间字段转换：将created、lastUpdated字段转换为标准时间类型，写入ES后会自动识别为日期格式，支持时间范围查询
  • 冗余字段清理：删除了原始的message字段以及Logstash默认生成的无关字段，降低存储占用
• 输出部分新增document_id配置，直接复用Mongo中的_id作为ES文档ID，避免重复导入时生成重复数据
• 输出的stdout指定了rubydebug编码，方便你在控制台查看解析后的字段结构，排查问题

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