如何通过Logstash将MongoDB嵌套JSON完整解析写入Elasticsearch

你当前遇到的解析失败核心原因是log_entry字段存储的不是标准JSON格式，而是Ruby Hash的序列化输出，包含=>赋值符、BSON::ObjectId()这类非JSON语法标识，因此直接使用JSON过滤器无法解析，Grok匹配嵌套结构容错率也极低。

方案1：使用Ruby过滤器兼容处理非标准格式（兼容性最好，无需升级插件） #

直接在filter段新增ruby过滤器，完成字符串修正、JSON解析操作，替换原有无效的json过滤器配置，完整filter配置如下：

filter {
    mutate  {
        rename => { "_id" => "mongo_id" }
        remove_field => ["host", "@version"]
    }
    # 仅对存在log_entry字段的异常文档做处理
    if [log_entry] {
        ruby {
            code => "
                # 拿到log_entry原始字符串
                raw_str = event.get('log_entry')
                return if raw_str.nil?
                # 替换Ruby Hash符号=>为JSON标准的:
                raw_str = raw_str.gsub('=>', ':')
                # 替换BSON::ObjectId('xxx')为字符串"xxx"
                raw_str = raw_str.gsub(/BSON::ObjectId\('([a-f0-9]+)'\)/, '"\1"')
                # 解析为JSON
                require 'json'
                begin
                    parsed_data = JSON.parse(raw_str)
                    # 把解析到的字段合并到事件根节点，也可以指定存到log子字段
                    parsed_data.each do |k, v|
                        event.set(k, v)
                    end
                    # 解析完成后删除冗余的log_entry字段
                    event.remove('log_entry')
                rescue JSON::ParserError => e
                    # 解析失败时打标记，方便后续排查
                    event.set('parse_error', e.message)
                end
            "
        }
    }
    # 可选：如果不需要解析后的_id字段可以直接删除
    mutate {
        remove_field => ["_id", "parse_error"]
    }
}

方案2：升级MongoDB输入插件直接解析BSON结构（性能最优） #

你当前使用的旧版本MongoDB输入插件没有自动处理BSON嵌套结构，升级插件后可在input段直接配置参数完成解析，无需后续复杂处理：

1. 升级插件命令：logstash-plugin install logstash-input-mongodb
2. 调整input配置：

input {
    mongodb {
        uri => "<mongouri>"
        placeholder_db_dir => "<path>"
        collection => "modules"
        batch_size => 5000
        # 新增以下两个参数
        bson_object_id_conversion => "string"
        flatten_structs => false # 保留嵌套结构不扁平化
    }
}

升级后嵌套的modules字段会直接生成，无需额外filter解析。

配置完成后先清空测试索引的旧数据，重新同步全量数据即可看到完整嵌套结构写入ES，如果需要对modules字段做嵌套查询，提前为索引设置nested类型mapping即可生效。

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