Elasticsearch：如何轻松安全地对实时 Elasticsearch 索引重新索引你的数据

news2025/1/15 6:30:53

在很多的时候，由于一些需求，我们不得不修改索引的映射，也即 mapping，这个时候我们需要重新索引（reindex）来把之前的数据索引到新的索引中。槽糕的是，我们的这个索引还在不断地收集实时数据，那么我们该如何处理这种情况呢？

比如，我们有这样的一个案例。假设你有一个名为 production_logs 的索引处于活动状态，这意味着它不断接收新数据。现在假设你想要以需要重新索（reindex）引该索引的所有数据的方式更新其映射：通常，当你想要更改现有字段的数据类型（例如从keyword 到 integer）时，会发生这种情况。

好的，现在你拥有的选项将取决于你首先如何设置索引。

你需要一个 index template

如果你没有自己创建索引，Elasticsearch 能够创建索引，这意味着如果你尝试索引 foo 索引中的某些数据，Elasticsearch 将创建它（如果它尚不存在）。

通过这样做，它使用称为动态映射的功能为这个新索引创建默认映射。

这就是你需要索引模板的原因！此功能允许你定义 Elasticsearch 自动创建的索引将获得的所有属性，包括其设置和映射。

因此，如果你的 production_logs 索引没有索引模板，仍然是时候创建它了。我们称它为 production_logs_template 并将此模板的模式设置为 production_logs* ，这意味着每次 Elasticearch 将自动创建名称与模式 production_logs* 匹配的索引时，它将应用该模板（请参阅完整文档）。

以下是索引模板创建请求的示例：

PUT _index_template/production_logs_template
{
  "index_patterns": [ "production_logs*"],
  "template": {
    "settings": {
      ...
    },
    "mappings": {
      ...
    }
  }
}

案例 1：你已经有一个 index alias

将真实索引隐藏在索引别名后面始终是一个好习惯。

它引入的抽象级别可帮助你在后台执行修改，而不会影响索引的用户（无论是人还是软件）。

因此，假设你获得了当前链接到要修改的 production_logs 索引的日志别名，并且你想要根据 production_logs_template 更新索引的映射。

这里的方法很简单：

更新索引模板以定义新映射
创建一个新索引，其名称与 production_logs* 匹配（例如 production_logs_1 ）
切换日志别名以指向这个新索引

POST _aliases
{
  "actions": [
    {
      "add": {
        "index": "production_logs_1",
        "alias": "logs"
      }
    },
    {
      "remove": {
        "index": "production_logs",
        "alias": "logs"
      }
    }
  ]
}

在上面我们的 add 及 remove 是一个原子操作，也即同时起作用。中间不会有其它的操作，这样保证在删除的同时，向我们的索引别名 logs 写入的数据能够正确地写入到新的索引中。

将数据从 production_logs 重新索引到 production_logs_1

POST _reindex?wait_for_completion=false
{
  "source": {
    "index": "production_logs"
  },
  "dest": {
    "index": "production_logs_1"
  }
}

此请求将返回一个 task_id，你可以使用它来监视重建索引过程的进度。

以下是你收到的响应示例：

{
  "task": "<task_id>"
}

你可以发出请求以利用任务管理 API：

GET _tasks/<task_id>?human

上述命令可以让我们知道任务的进度。就是这么简单！在重建索引过程结束时，你的 production_logs_1 索引将包含所有新旧数据，并具有正确的映射。 👏

案例2：你还没有一个 index alias

嗯，这会更难，但没有什么是不可能的，对吧？

从现在开始，我们仍然可以面临（至少）两种不同的情况：你是否使用摄入管道（ingest pipeline）。

在下文中，请记住我们已经创建了一个驱动 production_logs* 索引的索引模板。

案例 2.1：你正在使用一个 ingest pipeline

假设使用 production_logs_pipeline 将数据索引到 Elasticsearch，该管道处理任何传入事件，然后再将其索引到 production_logs 索引中。

以下是你想要更新实时索引的几个步骤。

首先，修改 production_logs_pipeline，在末尾添加 set processor。

{
  "set": {
    "field": "_index",
    "value": "{{{_index}}}_1"
  }
}

现在，所有通过此摄取管道的文档都将重定向到 production_logs_1 索引。

然后，你可以将所有数据从 production_logs 重新索引到名为 production_logs_orig 的新索引索引。

POST _reindex?wait_for_completion=false
{
  "source": {
    "index": "production_logs"
  },
  "dest": {
    "index": "production_logs_orig"
  }
}

这里请注意，索引 production_logs_1 和 production_logs_orig 都将根据索引模板 production_logs_template 创建！

重新索引任务完成后，你可以安全地删除 production_logs 索引（它的所有数据都已存储在 production_logs_orig 索引中）。

现在我们可以创建一个新的索引别名，它将被命名为 production_logs 并将同时针对 production_logs_orig 和 production_logs_1 索引，后者将是写入索引。

POST _aliases
{
  "actions": [
    {
      "add": {
        "index": "production_logs_orig",
        "alias": "production_logs"
      }
    },
    {
      "add": {
        "index": "production_logs_1",
        "alias": "production_logs",
        "is_write_index": true
      }
    }
  ]
}

最后，你需要返回到以前版本的 production_logs_pipeline，以便传入的数据从现在开始将使用新创建的 production_logs 别名。

作为最终结果，我们现在有一个索引别名指向两个具有预期映射的索引👏

💡 有可能再次从 production_logs_orig 重新索引到 production_logs_1 最后只有一个索引。

案例 2.2：索引是直接对索引进行的，没有摄取管道

在这种情况下，需要执行更多步骤，遗憾的是无法创建别名来替换原始索引，但你仍然可以将新映射应用于实时索引。

因此，首先你必须创建一个新的摄取管道，我们将其命名为 temp_pipeline：

PUT _ingest/pipeline/temp_pipeline
{
  "processors": [
    {
      "set": {
        "field": "_index",
        "value": "{{{_index}}}_1"
      }
    }
  ]
}

它只有一个处理器，其作用是将传入的文档重定向到一个新索引，该索引的名称是原始索引并以 _1 为后缀。

现在我们要表明任何进入 production_logs 索引的文档现在都应该使用这个新的摄取管道。我们可以在我们的索引上利用 index.default_pipeline 设置，这将允许我们应用管道，即使原始索引请求没有提到它。

PUT production_logs/_settings
{
  "index": {
    "default_pipeline": "temp_pipeline"
  }
}

从现在开始，每个传入的文档都将被重定向到 production_logs_1 索引，该索引将使用 production_logs_template 索引模板自动创建。

production_logs 索引不会获得任何新文档，因此我们可以将数据从该索引重新索引到一个名为 production_logs_orig 的新文档。

POST _reindex?wait_for_completion=false
{
  "source": {
    "index": "production_logs"
  },
  "dest": {
    "index": "production_logs_orig"
  }
}

和以前一样，我们可以使用任务管理 API来监控重建索引过程。

完成后，我们删除 production_logs，下一个传入文档到达后将立即重新创建！它将获得预期的映射（感谢索引模板）所以我们处于这种情况：

production_logs_orig 索引包含在我们开始操作之前索引的所有数据
production_logs_1 索引包含从迁移开始到删除之前的 production_logs 索引之间已编制索引的所有数据
production_logs 索引已重新创建并收集所有新数据（在如下的步骤中进行操作）

要回到单索引状态，我们只需将数据从 production_logs_orig 和 production_logs_1 重新索引到 production_logs 。

POST _reindex?wait_for_completion=false
{
  "source": {
    "index": "production_logs_1, production_logs_orig"
  },
  "dest": {
    "index": "production_logs"
  }
}

作为最终结果，我们将所有以前的数据（以及所有新数据）与新映射一起存储在 production_logs 索引中 👏