拉取式摄入

3.0 版本引入

这是一个实验性功能，不建议在生产环境中使用。有关该功能进展的更新或如果您想提供反馈，请在 OpenSearch 论坛上加入讨论。

拉取式摄取使 OpenSearch 能够从 Apache Kafka 或 Amazon Kinesis 等流式源摄取数据。与客户端通过 REST API 主动将数据推送到 OpenSearch 的传统摄取方法不同，拉取式摄取允许 OpenSearch 通过直接从流式源检索数据来控制数据流。这种方法提供了精确一次的摄取语义和原生反压处理，有助于防止在流量高峰期间服务器过载。

先决条件

在使用拉取式摄取之前，请确保满足以下先决条件：

• 使用命令 bin/opensearch-plugin install <plugin-name> 为您的流式源安装摄取插件。有关更多信息，请参阅附加插件。OpenSearch 支持以下摄取插件：
  • ingestion-kafka
  • ingestion-kinesis
• 启用段复制和远程支持存储。拉取式摄取与文档复制不兼容。
• 在索引创建期间配置拉取式摄取。您无法将现有的推送式索引转换为拉取式索引。

为拉取式摄取创建索引

要从流式源摄取数据，首先创建一个具有拉取式摄取设置的索引。以下请求创建一个从 Kafka topic 拉取数据的索引：

PUT /my-index
{
  "settings": {
    "ingestion_source": {
      "type": "kafka",
      "pointer.init.reset": "earliest",
      "param": {
        "topic": "test",
        "bootstrap_servers": "localhost:49353"
      }
    },
    "index.number_of_shards": 1,
    "index.number_of_replicas": 1,
    "index": {
      "replication.type": "SEGMENT"
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text"
      },
      "age": {
        "type": "integer"
      }
    }
  }
}

摄取源参数

ingestion_source 参数控制 OpenSearch 如何从流式源拉取数据。*轮询*是 OpenSearch 主动从流式源请求一批数据的操作。下表列出了 ingestion_source 支持的所有参数。

流位置

创建索引时，您可以通过在 ingestion_source 参数中配置 pointer.init.reset 和 pointer.init.reset.value 设置来指定 OpenSearch 应从流的何处开始读取。对于现有索引，OpenSearch 将从最后提交的位置恢复读取。

下表提供了有效的 pointer.init.reset 值及其对应的 pointer.init.reset.value 值。

流分区

使用分区流（例如 Kafka topic 或 Kinesis 分片）时，请注意流分区和 OpenSearch 分片之间的以下关系：

• OpenSearch 分片与流分区一一对应。
• 索引分片的数量必须大于或等于流分区的数量。
• 超出分区数量的额外分片将保持为空。
• 文档必须发送到同一分区才能成功更新。

使用拉取式摄取时，传统基于 REST API 的摄取将对该索引禁用。

更新错误策略

您可以使用更新设置 API 通过将 index.ingestion_source.error_strategy 设置为 DROP 或 BLOCK 来动态更新错误策略。

以下示例演示了如何更新错误策略：

PUT /my-index/_settings
{
  "index.ingestion_source.error_strategy": "DROP"
}

消息格式

为了由 OpenSearch 正确处理，流式源中的消息必须具有以下格式：

{"_id":"1", "_version":"1", "_source":{"name": "alice", "age": 30}, "_op_type": "index"}
{"_id":"2", "_version":"2", "_source":{"name": "alice", "age": 30}, "_op_type": "delete"}

流式源中的每个数据单元（Kafka 消息或 Kinesis 记录）必须包含以下字段，这些字段指定如何创建或修改 OpenSearch 文档。

拉取式摄取指标

拉取式摄取提供可用于监控摄取过程的指标。目前支持 polling_ingest_stats 指标，并在分片级别可用。

下表列出了可用的 polling_ingest_stats 指标。

要检索分片级别的拉取式摄取指标，请使用节点统计 API

GET /_nodes/stats/indices?level=shards&pretty

限制

使用拉取式摄取时，存在以下限制：

• 摄取管道与拉取式摄取不兼容。
• 不支持动态映射。
• 不支持索引翻转。
• 不支持操作监听器。

相关文章

• 基于拉取的摄取管理