如何在elasticsearch管道中指定文档版本？

Question

我目前使用一个接收节点管道,如下所示:

{
    "my-pipeline": {
        "description": "pipeline for my filebeat",
        "processors": [
            {
                "json": {
                    "field": "message",
                    "add_to_root": true,
                    "on_failure": [
                        {
                            "rename": {
                                "field": "message",
                                "target_field": "originalMessage",
                                "ignore_missing": true
                            }
                        },
                        {
                            "set": {
                                "field": "indexName",
                                "value": "pipeline-errors"
                            }
                        },
                        {
                            "set": {
                                "field": "indexType",
                                "value": "pipeline-error"
                            }
                        },
                        {
                            "rename": {
                                "field": "@timestamp",
                                "target_field": "errorTimestamp",
                                "ignore_missing": true
                            }
                        }
                    ]
                }
            },
            {
                "remove": {
                    "field": "@timestamp",
                    "ignore_failure": true
                }
            },
            {
                "remove": {
                    "field": "message",
                    "ignore_failure": true
                }
            },
            {
                "script": {
                    "inline": "ctx._index = ctx.indexName; ctx._type=ctx.indexType; if (ctx.docVersion != null) {ctx._version = ctx.docVersion; ctx._version_type='external'}"
                }
            },
            {
                "remove": {
                    "field": "indexName",
                    "ignore_failure": true
                }
            },
            {
                "remove": {
                    "field": "indexType",
                    "ignore_failure": true
                }
            }
        ]
    }
}

此管道仅用于取消由filebeat转发的日志.在脚本处理器中,我查找'indexName'和'indexType'字段,并分别将其分配给'_index'和'_type'.由于我需要考虑版本,因此日志中包含"版本"字段(但这是可选的,因为某些日志不包含该版本).

使用此管道触发:

org.elasticsearch.index.mapper.MapperParsingException: Cannot generate dynamic mappings of type [_version] for [_version]
    at org.elasticsearch.index.mapper.DocumentParser.createBuilderFromFieldType(DocumentParser.java:656) ~[elasticsearch-5.5.0.jar:5.5.0]
    at org.elasticsearch.index.mapper.DocumentParser.parseDynamicValue(DocumentParser.java:805) ~[elasticsearch-5.5.0.jar:5.5.0]

我到目前为止所尝试的内容(更新时间为09-16):

• 将字段名称替换为"docVersion"之类的内容,以确保它的关键字不会发生冲突.这也行不通
• 试图使用ctx._source.version,这会触发ScriptException [运行时错误]; 毕竟,请注意_index和_type值分别来自ctx.indexName和ctx.indexType
• 尝试在脚本上添加'version_type = external';我仍然得到如上所述的MapperParsingException;
• 尝试使用'version_type = external_gte',但我也得到了MapperParsingException

使用ingester节点管道时,如何在elasticsearch文档中指定/使用外部版本控制？如果通过管道的脚本处理器无法做到这一点,那么在使用filebeat-to-elasticsearch时使用外部版本的选项有哪些选择,以使文档的旧版本被拒绝？

更新10-24-2017 似乎这是当前elasticsearch版本不存在的功能(在我的情况下为5.6).根据代码中的检查,管道执行服务中的IndexRequest不包括对文档版本或版本类型的任何引用,因此默认为内部版本.也许这可以作为未来弹性搜索版本中的一项功能添加.

Answer 1

以下变量可通过 ctx 映射获得：_index、_type、_id、_version、_routing、_parent、_now 和 _source。您可以通过 ctx._source.field-name 访问字段的原始源。

看起来脚本正在尝试通过 ctx.version 访问名为“version”的文档字段，但该字段映射到 ctx._version。

内部文档值应作为 ctx._source.version 检索，您可以尝试一下吗？