使用 python 进行 Elastic Search 批量更新，如何使用新数据追加数组字段

Question

如何使用 python 中的bulkupdate 更新弹性搜索中的字段。我尝试了很多方法，但都出错了。在某些情况下，我收到文档丢失错误，如何同时更新和更新插入。而且附加到字段也不起作用。elasticsearch==7.9.1 是我在 python 中使用的包for i in range(0, length, steps):    end_index = length-1 if i+steps>length else i+steps    temp_list = test_data[i: end_index]    bulk_file = ''    actions = [{        "_index": "test-data",        "_opt_type":"update",        "_type": "test-test-data",        "_id": test_row ['testId'],        "doc":{"script": {                          "source": "ctx._source.DataIds.add(params.DataIds)",                          "lang": "painless",                          "params": {                              "DataIds":test_row ['DataIds']                          }                      }}        }        for test_row in temp_list    ]    helpers.bulk(es, actions)我得到的错误是这样的    {'update': {'_index': 'test-data', '_type': 'products', '_id': '333', 'status': 400, 'error': {'type': 'illegal_argument_exception', 'reason': 'failed to execute script', 'caused_by': {'type': 'script_exception', 'reason': 'runtime error', 'script_stack': ['ctx._source.dataIds.add(params.dataIds)', '    ^---- HERE'], 'script': 'if (ctx._source.dataIds == null) { ctx._source.dataIds = []; } ctx._source.dataIds.add(params.dataIds)', 'lang': 'painless', 'position': {'offset': 105, 'start': 71, 'end': 118}, 'caused_by': {'type': 'illegal_argument_exception', 'reason': 'dynamic method [java.lang.String, add/1] not found'}}}, 'data': {'upsert': {}, 'scripted_upsert': True, 'script': {'source': 'if (ctx._source.dataIds == null) { ctx._source.dataIds = []; } ctx._source.dataIds.add(params.dataIds)', 'lang': 'painless', 'params': {'cdataIds': 'set123'}}}}}])

Answer 1

upsert通过脚本的正确方法是不使用该部分，doc而仅使用该script部分。upsert如果您想在同一命令中更新插入和更新，您还需要该部分。事情是这样的：

actions = [{

    "_op_type":"update",

    "_index": "test-data",

    "_type": "test-test-data",

    "_id": test_row ['testId'],

    "upsert": {

       "DataIds": test_row ['DataIds']

    },

    "script": {

        "source": "ctx._source.DataIds.add(params.DataIds)",

        "lang": "painless",

        "params": {

           "DataIds":test_row ['DataIds']

        }

    }

} for test_row in temp_list

]

另一种方法是使用scripted_upsert

actions = [{

    "_op_type":"update",

    "_index": "test-data",

    "_type": "test-test-data",

    "_id": test_row ['testId'],

    "upsert": {},

    "scripted_upsert": true,

    "script": {

        "source": "if (ctx._source.DataIds == null) { ctx._source.DataIds = []; } ctx._source.DataIds.add(params.DataIds)",

        "lang": "painless",

        "params": {

           "DataIds":test_row ['DataIds']

        }

    }

} for test_row in temp_list

]