4.3 更新文档 (Update API)

4.3 更新文档 (Update API)

4.3 更新文档 (Update API)

Elasticsearch 提供了 Update API，允许你部分更新已存在的文档，而无需重新索引整个文档。这对于只需要修改文档中的几个字段的情况非常有用，可以提高效率。

4.3.1 Update API 的工作原理

Update API 的工作流程如下：

1. 检索文档： 首先，Elasticsearch 根据文档 ID 检索需要更新的文档。

2. 执行更新脚本或部分文档： 将检索到的文档传递给更新脚本（使用 painless 或其他脚本语言）或部分文档。脚本或部分文档定义了如何修改文档。

3. 重新索引： 脚本执行后，或者部分文档合并后，Elasticsearch 将重新索引更新后的文档。

这个过程是原子性的，确保并发更新不会导致数据损坏。

4.3.2 使用 Update API

Update API 的基本语法如下：

POST /{index_name}/_update/{document_id}
{
  "script": {
    "source": "{script_expression}",
    "lang": "{script_language}",
    "params": {
      "{parameter_name}": {parameter_value}
    }
  },
  "doc": {
    "{field_name}": {new_value},
    "{field_name}": {new_value}
  },
  "upsert": {
    "{field_name}": {value},
    "{field_name}": {value}
  }
}
​

• {index_name}: 索引名称。

• {document_id}: 要更新的文档 ID。

• script: 包含更新脚本的 JSON 对象（可选）。

  • source: 脚本的实际代码。

  • lang: 脚本语言（例如，painless）。

  • params: 传递给脚本的参数。

• doc: 包含要更新的字段和值的 JSON 对象（可选）。

• upsert: 如果文档不存在，则创建新文档的 JSON 对象（可选）。

可以选择使用 script 或 doc 来更新文档，但不能同时使用。upsert 只能与 script 或 doc 结合使用。

4.3.3 代码实践

以下是一些使用 Update API 的代码示例，使用Python Elasticsearch client。

1. 使用脚本更新文档：

假设我们有一个名为 products 的索引，其中包含以下文档：

{
  "name": "Elasticsearch Book",
  "price": 50,
  "quantity": 10
}
​

我们要将价格增加 10%，并将库存增加 5。

from elasticsearch import Elasticsearch
# 连接到 Elasticsearch
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
index_name = "products"
document_id = "1" # 假设文档ID是 1
# 更新文档
try:
    response = es.update(
        index=index_name,
        id=document_id,
        body={
            "script": {
                "source": """
                    ctx._source.price *= 1.1;
                    ctx._source.quantity += 5;
                """,
                "lang": "painless"
            }
        }
    )
    print("Update Response:", response)
except Exception as e:
    print("Error updating document:", e)
# 验证更新
try:
    get_response = es.get(index=index_name, id=document_id)
    print("Updated Document:", get_response['_source'])
except Exception as e:
    print("Error getting document:", e)
​

解释:

• 我们使用 script 参数来指定更新脚本。

• source 包含 Painless 脚本，用于更新 price 和 quantity 字段。

• ctx._source 用于访问文档的源数据。

2. 使用 doc 更新文档：

如果我们只想更新 quantity 字段，可以使用 doc 参数：

from elasticsearch import Elasticsearch
# 连接到 Elasticsearch
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
index_name = "products"
document_id = "1" # 假设文档ID是 1
# 更新文档
try:
    response = es.update(
        index=index_name,
        id=document_id,
        body={
            "doc": {
                "quantity": 15
            }
        }
    )
    print("Update Response:", response)
except Exception as e:
    print("Error updating document:", e)
# 验证更新
try:
    get_response = es.get(index=index_name, id=document_id)
    print("Updated Document:", get_response['_source'])
except Exception as e:
    print("Error getting document:", e)
​

解释:

• doc 参数包含要更新的字段和新值。

• 如果文档中不存在该字段，则会添加该字段。

• 如果字段已存在，则会更新其值。

3. 使用 upsert 创建文档：

如果文档不存在，我们可以使用 upsert 参数来创建新文档：

from elasticsearch import Elasticsearch
# 连接到 Elasticsearch
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
index_name = "products"
document_id = "2" # 假设文档ID是 2，且不存在
# 更新文档
try:
    response = es.update(
        index=index_name,
        id=document_id,
        body={
            "doc": {
                "name": "New Product",
                "price": 25,
                "quantity": 5
            },
            "upsert": {
                "name": "New Product",
                "price": 25,
                "quantity": 5
            }
        }
    )
    print("Update Response:", response)
except Exception as e:
    print("Error updating document:", e)
# 验证更新
try:
    get_response = es.get(index=index_name, id=document_id)
    print("Updated Document:", get_response['_source'])
except Exception as e:
    print("Error getting document:", e)
​

解释:

• doc 参数包含要更新的字段和新值。

• upsert 参数包含如果文档不存在时要创建的文档。

• 如果文档存在，则 doc 中的字段将更新现有文档。

• 如果文档不存在，则将创建 upsert 中定义的文档。

4. 使用 script 和 upsert 创建文档：

from elasticsearch import Elasticsearch
# 连接到 Elasticsearch
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
index_name = "products"
document_id = "3" # 假设文档ID是 3，且不存在
# 更新文档
try:
    response = es.update(
        index=index_name,
        id=document_id,
        body={
            "script": {
                "source": "ctx._source.quantity += params.count",
                "lang": "painless",
                "params": {
                    "count": 10
                }
            },
            "upsert": {
                "name": "Another Product",
                "price": 75,
                "quantity": 10
            }
        }
    )
    print("Update Response:", response)
except Exception as e:
    print("Error updating document:", e)
# 验证更新
try:
    get_response = es.get(index=index_name, id=document_id)
    print("Updated Document:", get_response['_source'])
except Exception as e:
    print("Error getting document:", e)
​

解释:

• 如果文档存在，脚本将增加 quantity 字段。

• 如果文档不存在，将使用 upsert 创建新文档，然后脚本将增加 quantity 字段。

4.3.4 Update API 的参数

Update API 支持许多参数，用于控制更新过程：

• retry_on_conflict: 指定在发生版本冲突时重试更新的次数。

• refresh: 控制更新后是否刷新索引，使其对搜索可见。

• _source: 控制是否返回更新后的文档源。

• if_seq_no 和 if_primary_term: 用于乐观并发控制，确保在更新时文档未被修改。

4.3.5 错误处理

Update API 可能会返回以下错误：

• DocumentMissingException: 如果文档不存在且未指定 upsert。

• VersionConflictEngineException: 如果发生版本冲突。

• ScriptException: 如果脚本执行失败。

4.3.6 流程图

流程图解释:

1. 客户端发起更新请求到 Elasticsearch。

2. Elasticsearch 接收到请求后，首先检查要更新的文档是否存在。

3. 如果文档存在，则执行更新脚本或使用 doc 更新文档。

4. 如果文档不存在，则检查是否指定了 upsert。

5. 如果指定了 upsert，则创建新文档。

6. 如果没有指定 upsert，则返回 DocumentMissingException 错误。

7. 更新或创建文档后，Elasticsearch 重新索引文档。

8. 最后，Elasticsearch 将更新结果返回给客户端。

9. 如果在任何阶段发生错误，Elasticsearch 将返回相应的错误信息。

4.3.7 总结

Update API 提供了一种有效的方法来部分更新 Elasticsearch 中的文档。你可以使用脚本或 doc 参数来指定更新，并使用 upsert 参数来创建新文档（如果文档不存在）。 理解 Update API 的工作原理以及可用的参数，可以帮助你更有效地管理 Elasticsearch 中的数据。 记住要处理可能发生的错误，并考虑使用乐观并发控制来确保数据一致性。