ConsumeElasticsearch

该 Processor 适用于 Elasticsearch JSON DSL 与 Elasticsearch 5.X 及以上 版本。它的设计目标是能够使用输入属性创建一个 JSON 查询，并以分页方式 对 Elasticsearch 集群执行。与 "restapi" bundle 中的所有 Processor 一样，它使用官方 Elastic 客户端 API，因此支持 leader 检测。

查询通过可用的方法之一在 Elasticsearch 中分页——"Scroll" 或 "Search After"（对于启用 XPack 的 Elasticsearch 7.10+，可以可选地 配合 "Point in Time"）。每页结果数量可以通过 size 属性控制。

结果会按要跟踪的字段排序，sort order 作为属性设置。

搜索结果和聚合结果可以拆分为多个 FlowFile。聚合结果只在顶层 拆分，因为嵌套聚合如果与父聚合分离，就会失去上下文（从而失去 其价值）。此外，所有页的结果可以合并为单个 FlowFile（但 Processor 在任何时刻都只把当前页的数据加载到内存）。

以下是为跟踪 "@timestamp" 字段而创建的查询示例：

{
  "query": {
    "size": 10000,
    "sort": {
      "@timestamp": "desc"
    },
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gt": "2023-09-01"
            }
          }
        }
      ]
    }
  }
}

可以在 query filter 属性中以 JSON 字符串添加额外的 "filter" 项，例如：

[
  {
    "term": {
      "department": "accounts"
    }
  },
  {
    "term": {
      "title.keyword": "Annual Report"
    }
  }
]

跨 Processor 执行的查询分页

该 Processor 按计划周期运行，以重复执行同一查询。一旦在 Elasticsearch 中发起分页查询，该 Processor 会持续获取同一查询的 结果，直到没有更多结果为止。之后，会使用相同的查询 JSON 发起 新的分页查询，但 "range" 过滤查询的 "gt" 字段会设为从上一次 结果中获取的最后一个值。

如果该 Processor 的结果是 "Combined"，分页查询会在单次调用内 持续运行，直到没有更多结果（之后该 Processor 会在下次调用时 启动新的分页查询）。如果结果是 "Split" 或 "Per Page"，该 Processor 的每次调用都会获取下一页结果，直到没有更多结果，或 分页查询在 Elasticsearch 中过期。

重置查询 / 清除 Processor 状态

Cluster State 用于跟踪该 Processor 中分页查询的进度。如果需要在 分页查询已开始之后完全重启查询或更改 Processor 配置，请务必在 停止该 Processor 之后、重启之前对其执行 "Clear State"。

注意，清除 Processor 状态会丢失对已获取文档的跟踪进度。在重启 该 Processor 之前，请用合适的值更新 "Initial Value"，以便从 之前的位置继续。

重复结果

该 Processor 不会尝试在查询之间去重。例如，如果同一查询运行两次 且结果（部分或全部）相同，两次的输出都会包含这些相同的结果。 当正在获取一页数据时 Lynxus 主节点（Primary Node）发生变化， 或者 Processor 状态被清除后重启，都可能发生这种情况。