Jak określić wersję dokumentu w potoku elasticsearch?

Question

Używam obecnie węzła rurociągu łykać który wygląda tak:

{ 
    "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 
       } 
      } 
     ] 
    } 
} 

Rurociąg ten służy po prostu unbox dziennik przekazane przez filebeat. W procesorze skryptów szukam pól "indexName" i "indexType" i przypisuję je odpowiednio do "_index" i "_typ". Ponieważ muszę wziąć wersję pod uwagę, w dzienniku znajduje się pole "wersja" (ale jest to opcjonalne, ponieważ niektóre dzienniki nie zawierają wersji).

Stosując ten rurociąg wyzwala:

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] 

co już próbowałem tak daleko (aktualizowane 09-16):

• Wymieniłem nazwę pola, aby coś jak '' tylko do DocVersion być pewny, że to nie koliduje, jeśli jest to słowo kluczowe. To nie działa zbyt
• Próbowałem użyć ctx._source.version, to spowodowałoby wyjątek ScriptException [błąd wykonania]; w końcu zauważ, że wartości _index i _typ pochodzą z ctx.indexName i ctx.indexType odpowiednio
• Próbowałem dodać "version_type = external" również w skrypcie, nadal otrzymuję MapperParsingException jak wyżej;
• spróbował wykorzystać „version_type = external_gte” ale mam MapperParsingException także

Jak określić/używać wersjonowanie dokumentów elasticsearch zewnętrznych w przypadku korzystania ingester rurociągów węzła? jeśli nie jest to możliwe za pomocą procesora skryptów rurociągów, jakie są opcje używania wersji zewnętrznej podczas pracy z funkcją Filebeat-to-Elasticsearch w taki sposób, że starsza wersja dokumentu zostanie odrzucona?

Aktualizacja 24.10.2017 Wydaje się, że jest to funkcja, która nie istnieje w obecnej wersji elasticsearch (5.6 w moim przypadku). Jak na sprawdzenie w code, IndexRequest w usłudze wykonania potoku nie zawiera żadnych odniesień do wersji dokumentu ani typu wersji, a więc domyślnie do wersji wewnętrznej. Być może można to dodać jako funkcję w przyszłych wydaniach elastycznych.

Answer 1

Następujące zmienne są dostępne na mapie ctx: _indeks, _typ, _id, _wersja, _routing, _parent, _now i _source. Możesz uzyskać dostęp do oryginalnego źródła dla pola jako ctx._source.field-name.

Wygląda na to, że skrypt próbuje uzyskać dostęp do pola dokumentu o nazwie "wersja" przez ctx.version, ale mapuje to do pliku ctx._version.

Wewnętrzna wartość doc powinna zostać pobrana jako ctx._source.version, możesz spróbować?