Elasticsearch заполняет два поля в одно новое поле после расчетов

Question

Вопрос. Мне поручено исследовать, как заполнять данные в Elasticsearch. До сих пор немного пуст. Основной смысл:

Примечания: Все документы хранятся под ежедневными индексами с ~ 200 тыс. Документов в день.

• Мне нужно иметь возможность переиндексации данных за 60 дней.
• Мне нужно взять два поля для каждого документа payload.time_sec и payload.time_nanosec, взять там значения и выполнить некоторую математику (time_sec * 10 ** 9 + time_nanosec), а затем вернуть это как одно поле в reindexed документ

Я смотрю на API документации Python с объемными помощниками: http://elasticsearch-py.readthedocs.io/en/master/helpers.html

Но мне интересно, если это вообще возможно.

Мои мысли должны были использовать: Массовые помощники, чтобы вытащить идентификатор прокрутки (массив _update?), Перебирать каждый идентификатор документа, извлекать эти данные из двух полей для каждой док-станции, выполнять математику и завершать обновление запрос с новыми данными поля.

Кто-нибудь это сделал? Может, что-то с серьезным сценарием?

Спасибо!

Answer 1

Объемные помощникам вытащить ID прокрутки (насыпной _update?), Перебирать каждый Идентификатор документа, тянуть, что данные из двух полей для каждого дока, сделать математику, и завершить запрос на обновление с новым полем данные.

В принципе, да:

• использование /_search?scroll для извлечения документации
• выполнить операцию
• отправить /_bulk запросы на обновление

Другие варианты:

• use the /_reindex API
  Наверное, не так хорошо, если вы не хотите, чтобы создать новый индекс
• use the /_update_by_query API

И поддержка сценариев, которые, если я правильно понял, пустошь быть идеальным выбором, потому что ваше обновление не зависят от внешних факторов, поэтому это может быть сделано непосредственно на сервере.

Answer 2

Вот где я нахожусь (примерно):

Ive работает с Python и объемными хелперов и до сих пор нахожусь здесь:

doc = helpers.scan(es, query={ 
"query": { 
"match_all": {} 

}, 
"size":1000 
},index=INDEX, scroll='5m', raise_on_error=False) 


    for x in doc: 
x['_index'] = NEW_INDEX 
try: 
    time_sec = x['_source']['payload']['time_sec'] 
    time_nanosec=x['_source']['payload']['time_nanosec'] 
    duration = (time_sec * 10**9) + time_nanosec 
except KeyError: pass 

count = count + 1 

x['_source']['payload']['duration'] = duration 
new_index_data.append(x) 

helpers.bulk(es,new_index_data) 

Отсюда я просто используя основную помощник python для вставки в новый индекс. Однако я буду экспериментировать с изменением и тестированием с массовым обновлением до существующего индекса.

Это похоже на правильный подход?