elasticsearch-ingest

Elasticsearch 文件导入

将本地数据文件加载到 Elasticsearch：先将其转换为批量 NDJSON，在类型重要时创建具有正确映射的索引，再批量索引文档并验证结果。

环境配置

此技能通过 elastic CLI 执行 Elasticsearch 操作。如果未安装 elastic CLI，请告知用户需要安装它。 不要猜测凭据、直接调用 HTTP API 或尝试其他变通方法。

此技能以 HTTP 简写形式引用操作（例如，GET /、GET /_cat/indices、GET /{index}/_mapping、 GET /{index}/_settings/index.mode、POST /_query）。本文档末尾的 操作 表 列出了每种简写对应的 elastic CLI 命令——始终使用 CLI，而不要直接调用 HTTP API。

范围

此技能涵盖通过 POST /_bulk 加载 文件 → 索引。它不使用 Logstash、Filebeat、Elastic Agent、 Node.js 导入工具或其他边车管道。要在现有索引之间复制文档，请使用索引到索引的重新索引，而不是重新解析源文件。

支持的源数据形态：

源数据形态	示例	批量导入要求
带标题行的 CSV	id,name,age,... 后接数据行	将标题行解析为字段名；每个数据行输出一条操作行和一个 JSON 对象
JSON 数组文件	[{"a":1},{"a":2}]	拆分为逐文档行——绝不能将原始数组作为单个文档批量加载
NDJSON / JSON Lines	每行一个 JSON 对象	如果缺少操作行，可选择添加；否则可直接用于批量导入

Parquet、Arrow 和其他二进制列式格式不在适用范围内，除非用户先将其转换为 CSV 或 JSON。

流程

1. 确认连接。 调用 GET /。如果调用失败，请停止操作，并先解决 CLI 配置问题，再读取文件或更改集群状态。

2. 检查源文件并判断其格式。 打开文件（或查看前几行），并判断其格式：

   • CSV — 第一行为逗号分隔的表头，后续各行为记录。统计数据行数（不包括表头）——加载后需报告此数量。
   • JSON 数组 — 文件以 [ 开头，包含对象数组。统计数组元素数量——每个元素会成为一份带索引的文档，而不是一份。
   • NDJSON — 每行一个 JSON 值；各行交替包含操作元数据和文档源数据，或者每行都是文档，但仍需在前面添加操作行。

   转换方式：根据 NDJSON Bulk Format 选择。绝不要将原始 CSV 文本或原始 JSON 数组作为请求体发送到 POST /_bulk。

3. 选择目标索引名称。 使用用户提供的名称，或根据文件名提出一个小写名称。索引名称必须为小写，不能包含空格或 /，且不应以 -、_ 或 + 开头。

4. 判断是否需要显式映射。 如果索引可能已存在，调用 GET /{index}/_mapping。

   在批量加载前创建显式映射的情况：

   • CSV 列包含必须能以类型化字段进行查询的数字、日期或布尔值（不能是纯文本）。
   • 用户要求使用可用的列类型或便于聚合的字段。
   • 之前的加载将所有内容都索引为 text/keyword 字符串，需要纠正。

   如果所有字段都可以保留为类似字符串的类型，且用户未指定类型，则首次批量摄取时动态映射可能足够——但对于 CSV，除非用户明确接受所有字段均为字符串类型，否则优先使用显式映射。

   请阅读 Mapping Design for Ingest 了解类型选择。如果索引已存在但类型错误，请先询问用户，再调用 DELETE /{index} 并重新创建。

5. 按需创建索引。 如果第 4 步要求使用显式类型（或索引不存在），请在批量加载前调用 PUT /{index}，并附带 mappings 块。不要依赖动态映射从 CSV 字符串单元格推断 long、date 或 boolean——动态映射通常会将含义不明确的字符串映射为 text，并添加 .keyword 子字段。

6. 将文件转换为 bulk NDJSON。 写入一个临时 NDJSON 文件，每份文档占两行：

   • 第 1 行 — 操作元数据，例如 {"index":{"_index":"<index>"}}（仅当用户要求稳定 ID 时才添加 "_id"）。
   • 第 2 行 — 字段类型正确的文档 JSON（数字使用 JSON 数字，布尔值使用 true/false，日期使用 ISO-8601 字符串，例如 2023-01-15）。

   对于 CSV，将表头行映射为 JSON 字段名，并将单元格值转换为与第 5 步中映射相符的 JSON 类型。对于 JSON 数组，遍历每个数组元素，并生成操作行 + 对象行。请参阅 NDJSON Bulk Format 中的示例。

7. 批量索引文档。 使用第 6 步生成的 NDJSON 文件调用 POST /_bulk。检查响应：如果 errors 为 true，请读取各项的 error 对象，修复映射或文档问题，并在修复后重试失败的项目。不能仅凭退出代码为零就认定操作成功。

8. 验证结果。 始终确认加载结果——绝不能仅根据检查文件报告数量。

   • 调用 GET /{index}/_count，并与第 2 步统计的预期行数/元素数比较。
   • 如果类型化列很重要，调用 GET /{index}/_mapping，确认 age 等字段为数值类型（long / integer）、日期为 date、布尔值为 boolean，而不是 text。

   报告已验证的文档数量，并在相关时报告已确认的字段类型。如果数量或映射检查未通过，请参阅 Troubleshooting。

指南

• 仅限批量操作。 所有文件加载都通过 POST /_bulk 和 NDJSON 操作行完成——批量文件不要循环执行单文档 PUT，不要用摄取管道替代客户端 CSV 解析，也不要直接提交未经处理的源文件。
• 必须拆分 JSON 数组。 将一个含 4 个元素的数组作为单个文档批量加载，得到的计数是 1；正确加载后，计数应为 4。
• CSV 表头即为模式。 第一行 CSV 数据定义字段；之后每行数据都是一个文档。一个包含 1 行表头和 5 行数据的文件，摄取后报告的计数必须为 5。
• 在文档 JSON 中进行类型转换。 CSV 单元格会以字符串形式传入；当映射声明了 long、date 或 boolean 时，应在批量请求正文中输出 JSON 数字、ISO 日期字符串和布尔字面量——不要依赖 Elasticsearch 在动态映射已将类型定为 text 后，再从带引号的 CSV 字符串推断类型。
• 对于有类型要求的 CSV，优先使用显式映射。 先通过 PUT /{index} 创建索引，可避免静默地将所有内容都按文本索引，从而导致范围查询和聚合失效。
• 幂等地重新加载。 重新加载到现有索引时，删除数据前先询问用户。未指定 _id 时，重复的批量 index 操作会追加新文档。

示例

包含类型化列的 CSV

源文件（users.csv — 表头 + 5 行数据）：

id,name,age,signup_date,active
1,Ada Lovelace,36,2023-01-15,true

使用显式类型创建索引，将数据行转换为 NDJSON（5 行数据对应 5 组操作行 + 文档行），执行批量加载，然后验证计数为 5 且映射类型正确。完整操作指南： 摄取映射设计 和 NDJSON 批量格式。

JSON 数组文件

源文件（events.json）：

[
  { "event_id": "e-1", "type": "login", "user_id": 1, "value": 12.5 },
  { "event_id": "e-2", "type": "logout", "user_id": 1, "value": 0.0 }
]

将数组中的 4 个元素分别转换为 4 组批量操作行（不要将整个数组转换为一组）。验证 GET /{index}/_count 返回 4。参见 NDJSON 批量格式。

已准备好的 NDJSON

如果文件中操作行和文档行交替出现，请先验证格式并确认目标索引和映射，然后直接将文件传给 POST /_bulk。

不适用场景

• 持续或流式摄取 — 使用 Elastic Agent 或 Beats 收集日志和指标。
• 复杂的丰富化管道 — 单独设计服务端摄取管道；此技能仍会在加载前于客户端将文件转换为批量 NDJSON。
• 索引间复制或映射迁移 — 在索引之间执行重新索引，而不是导出到文件。
• 超大的二进制列式文件 — 先离线转换为 CSV 或 JSON，再按照此技能操作。

参考资料

• NDJSON 批量格式 — CSV 和 JSON 数组转换、操作行语法、批次大小
• 摄取映射设计 — CSV 类型的显式映射、类似 eval 的模式
• 故障排除 — 计数错误、数值被映射为文本、批量项目错误

操作

HTTP API（简写）	elastic CLI 命令
GET /	elastic es info
PUT /{index}	elastic es indices create --index '<index>' --mappings '<json>'
DELETE /{index}	elastic es indices delete --index '<index>'
POST /_bulk	elastic es bulk --index '<index>' --input-file '<ndjson-path>'
GET /{index}/_count	elastic es count --index '<index>'
GET /{index}/_mapping	elastic es indices get-mapping --index '<index>'