PutElasticsearchJson

该 Processor 用于访问 Elasticsearch Bulk API。它支持按每个 FlowFile 配置批量操作，这正是它与 PutElasticsearchRecord 的区别。

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

该 Processor 会为每（批）个 FlowFile 构建一个 Elasticsearch Bulk API 请求体。应当注意把 FlowFile 批量为大小合适的块，避免 Lynxus 耗尽内存，同时确保发送给 Elasticsearch 的请求不会过大。 发生失败时，该 Processor 可以把失败的 FlowFile 路由到错误队列， 使只有失败的 FlowFile 可以在下游处理或重放。

index、operation 和（可选的）type 字段配置了默认值。ID（当 operation 为 "index" 时可选）可以设置为 FlowFile 的 Attribute。

字段路径模式（Field Path Mode）

Identifier Field、Index Field 和 Timestamp Field 属性 各自命名一个文档中的字段，其值会被提取（分别用作文档 ID、索引名 或 @timestamp）。Field Path Mode 属性控制这些值的解释方式：

• Literal Field Name（默认）— 每个值都是顶层字段的确切名称。 / 或 \ 只是字段名中的普通字符，因此像 @metadata/id 这样的值 匹配名称恰好为 @metadata/id 的顶层字段。
• Nested Field Path — 每个值是以 / 分隔的嵌套对象路径， 如下所述。

Literal Field Name（默认）

对于文档：

{
  "@metadata/id": "abc",
  "message": "Hello, world"
}

Identifier Field 为 @metadata/id 会选中 abc — 即名称恰好为 @metadata/id 的顶层字段。同样的值对于 {"@metadata": {"id": "abc"}} 不会选中任何内容，因为不存在名称恰好为该值的顶层字段。

Nested Field Path

本节其余部分适用于 Field Path Mode 设为 Nested Field Path 的情况。对于文档：

{
  "@metadata": {
    "id": "abc",
    "index": "my-index"
  },
  "message": "Hello, world"
}

• Identifier Field 路径 @metadata/id 选中 abc
• Index Field 路径 @metadata/index 选中 my-index

名称中包含斜杠的字段

由于 / 分隔路径段，名称中确实包含 / 的字段必须把该斜杠转义为 \/，字面反斜杠必须转义为 \\。例如，对于文档 {"a/b": "abc"}，路径 a\/b 选中名为 a/b 的顶层字段。

移除与裁剪

当对应的 Retain 属性设为 false 时，字段在提取值之后会从 文档正文中移除。对于嵌套路径，因移除而变空的父对象也会被裁剪。 例如，从 {"@metadata": {"id": "abc"}, "message": "Hello, world"} 中提取并移除 @metadata/id 后，剩下 {"message": "Hello, world"}， 已为空的 @metadata 对象被移除。

动态模板（Dynamic Templates）

Index 和 Create 操作可以使用动态模板。Dynamic Templates 属性 必须可解析为 JSON 对象。

示例 - 带动态模板的 Index

{
  "message": "Hello, world"
}

下面的 Dynamic Templates 属性可解析：

{
  "message": "keyword_lower"
}

会创建如下的 Elasticsearch action：

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

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

Update/Upsert 脚本

Update 和 Upsert 操作可以使用脚本。脚本必须包含 Elasticsearch 要求的所有元素，例如 source 和 lang。Script 属性必须可解析为 JSON 对象。

如果为 upsert 定义了脚本，FlowFile 内容会用作 Elasticsearch action 中的 upsert 字段。如果没有定义脚本，FlowFile 内容会用作 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
  }
}

会创建如下的 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 发送与为该 FlowFile 定义的索引操作相关的已知 Bulk action 字段，Elasticsearch 会拒绝索引操作与 Bulk action 字段的无效组合。

示例 - 带冲突重试的 Update

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

使用下面的动态属性：

• BULK:retry_on_conflict = 3

会创建如下的 Elasticsearch action：

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

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

Index 操作

"operation" 的有效值为：

• create
• delete
• index
• update