PutElasticsearchRecord

该 Processor 用于访问 Elasticsearch Bulk API。它支持按每条记录 配置批量操作，这正是它与 PutElasticsearchJson 的区别。例如，可以 先定义多条索引文档的命令，然后按需对同一索引或其他索引执行 delete、create 和 update 操作。

作为 Elasticsearch REST API bundle 的一部分，它使用一个 Controller Service 管理连接信息，该 Controller Service 构建在 官方 Elasticsearch 客户端 API 之上，提供了诸如针对集群的自动 master 检测等其他 bundle 不具备的功能。

该 Processor 会为每个记录集构建一个 Elasticsearch Bulk API 请求体。应当注意把记录集拆分为大小合适的块，避免 Lynxus 耗尽 内存，同时确保发送给 Elasticsearch 的请求不会过大。发生失败时， 该 Processor 可以把失败的记录写入输出 record writer，使只有失败的 记录可以在下游处理或重放。

逐记录操作（Per-Record Actions）

index、operation 和（可选的）type 字段配置了默认值，可以通过 记录路径（record path）操作覆盖，从记录集中查找 index 或 type 值。 ID 和操作类型（create、index、update、upsert 或 delete）也可以 用类似方式从记录集中提取。"@timestamp" 字段可以使用默认值添加， 也可以从记录集中提取。当文档要索引到 Elasticsearch 数据流 （Data Stream）时，这会很有用。

示例 - 逐记录操作

以下示例文档演示了所有这些功能：

{
  "metadata": {
    "id": "12345",
    "index": "test",
    "type": "message",
    "operation": "index"
  },
  "message": "Hello, world",
  "from": "john.smith",
  "ts": "2021-12-03'T'14:00:00.000Z"
}

{
  "metadata": {
    "id": "12345",
    "index": "test",
    "type": "message",
    "operation": "delete"
  }
}

下面的记录路径操作会提取相关数据：

• /metadata/id
• /metadata/index
• metadata/type
• metadata/operation
• /ts

动态模板（Dynamic Templates）

Index 和 Create 操作可以使用来自记录的动态模板，可以配置 记录路径操作从记录集中查找动态模板。记录中的动态模板 字段必须是 Map、子 Record，或者可解析为 JSON 对象的字符串。

示例 - 带动态模板的 Index

{
  "message": "Hello, world",
  "dynamic_templates": "{\"message\": \"keyword_lower\"}"
}

下面的记录路径操作会提取相关的动态模板：

• /dynamic_templates

会创建如下的 Elasticsearch action：

{
  "index": {
    "_id": "1",
    "_index": "test",
    "dynamic_templates": {
      "message": "keyword_lower"
    }
  }
}

{
  "doc": {
    "message": "Hello, world"
  }
}

Update/Upsert 脚本

Update 和 Upsert 操作可以使用来自记录的脚本，可以配置记录路径 操作从记录集中查找脚本。脚本必须包含 Elasticsearch 要求的所有 元素，例如 source 和 lang。记录中的脚本字段必须是 Map、子 Record，或者可解析为 JSON 对象的字符串。

如果为 upsert 定义了脚本，记录中剩余的字段会用作 Elasticsearch action 中的 upsert 字段。如果没有定义脚本， 所有记录字段都会用作 update 的 doc（对于 upsert 操作则为 doc_as_upsert）。

示例 - 不带脚本的 Update

{
  "message": "Hello, world",
  "from": "john.smith"
}

会创建如下的 Elasticsearch action：

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

示例 - 带脚本的 Upsert

{
  "counter": 1,
  "script": {
    "source": "ctx._source.counter += params.param1",
    "lang": "painless",
    "params": {
      "param1": 1
    }
  }
}

下面的记录路径操作会提取相关脚本：

• /script

会创建如下的 Elasticsearch action：

{
  "update": {
    "_id": "1",
    "_index": "test"
  }
}

{
  "script": {
    "source": "ctx._source.counter += params.param1",
    "lang": "painless",
    "params": {
      "param1": 1
    }
  },
  "upsert": {
    "counter": 1
  }
}

Bulk Action 头字段

可以在 Processor 上定义带 BULK: 前缀的动态属性。动态属性的 值是从记录集中查找字段值的记录路径操作。用户必须确保只向 Elasticsearch 发送与为该记录定义的索引操作相关的已知 Bulk action 字段，Elasticsearch 会拒绝索引操作与 Bulk action 字段的无效组合。

示例 - 带冲突重试的 Update

{
  "message": "Hello, world",
  "from": "john.smith",
  "retry": 3
}

下面的动态属性和记录路径操作会提取相关字段：

• BULK:retry_on_conflict = /retry

会创建如下的 Elasticsearch action：

{
  "update": {
    "_id": "1",
    "_index": "test",
    "retry_on_conflict": 3
  }
}

{
  "doc": {
    "message": "Hello, world",
    "from": "john.smith"
  }
}

Index 操作

"operation" 的有效值为：

• create
• delete
• index
• update