如何将S3中的JSON文件对象作为单个文档创建Elasticsearch索引

S3 同步至 Elasticsearch 单文件单文档配置方案 #

问题根因 #

当前配置把单个JSON拆成多条文档写入，是因为Logstash S3输入插件默认以换行符作为事件分割边界。你存在S3里的JSON是带换行缩进的格式化内容，插件会把每一行识别为独立事件，再配合input端直接配置的json codec逐行解析，最终把完整对象拆成了多个键值对级别的零散文档。

修正后管道配置 #

input {
    s3 {
        access_key_id => "MY_KEY"
        secret_access_key => "MY_SECRET"
        bucket => "sthreetoes"
        region => "ap-south-1"
        # 替换原json codec，用multiline合并单个文件的所有内容
        codec => multiline {
            pattern => "^\{"
            negate => true
            what => "previous"
        }
        # 配置消费进度记录路径，避免重启后重复同步文件
        sincedb_path => "/var/lib/logstash/s3_sync_sincedb"
    }
}

filter {
    # 对合并完成的完整JSON字符串做解析
    json {
        source => "message"
        # 解析完成后删除冗余的原始字段
        remove_field => ["message", "@version", "tags"]
    }
}

output {
    elasticsearch {
        hosts => "http://elasticsearch:9200"
        index => "test-data"
        # 用JSON自带的id作为ES文档ID，避免重复导入产生重复数据
        document_id => "%{id}"
    }
}

关键配置说明 #

• multiline codec规则：将所有不以{开头的行全部归并到上一个事件，只要你的单个S3文件对应一个以{开头的JSON对象，不管格式缩进有多少换行，都会被拼接成完整的JSON字符串，不会拆分为多个事件。
• JSON解析逻辑移到filter段执行，确保拿到完整的文件内容后再做结构解析，从根源避免逐行解析导致的结构拆分。
• 新增的document_id配置可选，如果你有重复同步的需求，配置后同一个id的文档会被覆盖，不会产生重复数据。
• sincedb_path建议配置，Logstash会把已经消费过的S3文件位置记录在这个路径下，服务重启后不会从头拉取所有文件重复同步。

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