Elasticsearch 计算后将两个字段回填为一个新字段

Question

问题。我的任务是研究如何在 Elasticsearch 中回填数据。到目前为止，有点空。基本要点是：

注意：所有文档都存储在每日索引下，每天大约有 200k 文档。

• 我需要能够重新索引大约 60 天的数据。
• 我需要为每个文档的 payload.time_sec 和 payload.time_nanosec 取两个字段，取其中的值并对它们进行一些数学运算（time_sec * 10**9 + time_nanosec），然后将其作为单个字段返回到重新索引的文件

我正在查看带有批量助手的 Python API 文档： http://elasticsearch-py.readthedocs.io/en/master/helpers.html

但我想知道这是否可能。

我的想法是使用： 批量帮助器提取滚动 ID（批量 _update？），遍历每个文档 ID，从每个停靠栏的两个字段中提取数据，进行数学运算，并使用新字段数据完成更新请求。

有人做过吗？也许有一个 groovy 脚本的东西？

谢谢！

Answer 1

批量帮助器提取滚动 ID（批量 _update？），遍历每个文档 ID，从每个停靠点的两个字段中提取数据，进行数学运算，并使用新字段数据完成更新请求。

基本上是的：

• 使用/_search?scroll 获取文档
• 执行您的操作
• 发送/_bulk更新请求

其他选项有：

• use the /_reindex API
  如果您不想创建新索引，可能不太好
• use the /_update_by_query API

两者都支持脚本，如果我理解正确，这将是完美的选择，因为您的更新不依赖于外部因素，因此也可以直接在服务器内完成。

Answer 2

这是我所在的位置（大致）：

我一直在使用 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 帮助程序插入新索引。不过，我将尝试对现有索引进行批量更新来进行更改和测试。

这看起来是一种正确的方法？