elasticsearch-file-ingest

Elasticsearch 文件导入

将大型数据文件（NDJSON、CSV、Parquet、Arrow IPC）以流式方式导入并转换为 Elasticsearch 数据。

功能与使用场景

• 流式处理：处理大型文件而不会耗尽内存
• 高吞吐量：在普通硬件上每秒处理 50k+ 个文档
• 格式：NDJSON、CSV、Parquet、Arrow IPC
• 转换：在导入过程中应用自定义 JavaScript 转换（丰富数据、拆分、筛选）
• 批处理：导入符合模式的多个文件（例如，logs/*.json）
• 文档拆分：将一个源文档转换为多个目标文档

先决条件

• 可访问 Elasticsearch 8.x 或 9.x（本地或远程）
• 已安装 Node.js 22+

设置

此技能是自包含的。scripts/ 文件夹和 package.json 位于此技能目录中。请从此目录运行所有命令。 引用其他位置的数据文件时，请使用绝对路径。

首次使用前，请安装依赖项：

npm install

环境配置

用户只能通过环境变量配置 Elasticsearch 连接。绝不要将凭据作为命令行参数传递。如果测试失败，请向用户输出以下设置选项，然后停止。连接测试成功前，不要继续导入。

选项 1：Elastic Cloud（生产环境推荐）

export ELASTICSEARCH_CLOUD_ID="<your-cloud-id>"
export ELASTICSEARCH_API_KEY="<your-api-key>"

选项 2：使用 API 密钥的直接 URL

export ELASTICSEARCH_URL="https://elasticsearch:9200"
export ELASTICSEARCH_API_KEY="<your-api-key>"

选项 3：基本身份验证

export ELASTICSEARCH_URL="https://elasticsearch:9200"
export ELASTICSEARCH_USERNAME="<your-username>"
export ELASTICSEARCH_PASSWORD="<your-password>"

选项 4：本地开发

如需本地开发和测试，请参阅 在本地运行 Elasticsearch 以启动 Elasticsearch 和 Kibana。设置完成后，按照上方选项 2 或选项 3 所示导出连接变量（URL 和 API 密钥或凭据）。

可选：跳过 TLS 验证（仅限开发环境）

export ELASTICSEARCH_INSECURE="true"

测试连接

导入数据前，请验证 Elasticsearch 连接：

node scripts/ingest.js test

务必先运行此测试。如果测试失败，请先解决连接问题再继续。

示例

导入 JSON 文件

node scripts/ingest.js ingest --file /absolute/path/to/data.json --target my-index

通过 stdin 流式处理 NDJSON/CSV

# NDJSON
cat /absolute/path/to/data.ndjson | node scripts/ingest.js ingest --stdin --target my-index

# CSV
cat /absolute/path/to/data.csv | node scripts/ingest.js ingest --stdin --source-format csv --target my-index

直接导入 CSV

node scripts/ingest.js ingest --file /absolute/path/to/users.csv --source-format csv --target users

直接导入 Parquet

node scripts/ingest.js ingest --file /absolute/path/to/users.parquet --source-format parquet --target users

直接导入 Arrow IPC

node scripts/ingest.js ingest --file /absolute/path/to/users.arrow --source-format arrow --target users

使用解析器选项导入 CSV

# csv-options.json
# {
#   "columns": true,
#   "delimiter": ";",
#   "trim": true
# }

node scripts/ingest.js ingest --file /absolute/path/to/users.csv --source-format csv --csv-options csv-options.json --target users

从 CSV 推断映射/管道

使用 --infer-mappings 时，不要将其与 --source-format csv 一起使用。推断会向 Elasticsearch 的 _text_structure/find_structure 端点发送原始样本，该端点会返回映射以及包含 CSV 处理器的导入管道。如果同时设置 --source-format csv，CSV 将在客户端和服务端各解析一次，导致索引为空。请让 --infer-mappings 负责所有处理：

node scripts/ingest.js ingest --file /absolute/path/to/users.csv --infer-mappings --target users

使用选项推断映射

# infer-options.json
# {
#   "sampleBytes": 200000,
#   "lines_to_sample": 2000
# }

node scripts/ingest.js ingest --file /absolute/path/to/users.csv --infer-mappings --infer-mappings-options infer-options.json --target users

使用自定义映射导入

node scripts/ingest.js ingest --file /absolute/path/to/data.json --target my-index --mappings mappings.json

使用转换导入

node scripts/ingest.js ingest --file /absolute/path/to/data.json --target my-index --transform transform.js

命令参考

必需选项

--target <index>         # Target index name

源选项（选择一项）

--file <path>            # Source file (supports wildcards, e.g., logs/*.json)
--stdin                  # Read NDJSON/CSV from stdin

索引配置

--mappings <file.json>          # Mappings file
--infer-mappings                # Infer mappings/pipeline from file/stream (do NOT combine with --source-format)
--infer-mappings-options <file> # Options for inference (JSON file)
--delete-index                  # Delete target index if exists
--pipeline <name>               # Ingest pipeline name

处理

--transform <file.js>    # Transform function (export as default or module.exports)
--source-format <fmt>    # Source format: ndjson|csv|parquet|arrow (default: ndjson)
--csv-options <file>     # CSV parser options (JSON file)
--skip-header            # Skip first line (e.g., CSV header)

性能

--buffer-size <kb>       # Buffer size in KB (default: 5120)
--total-docs <n>         # Total docs for progress bar (file/stream)
--stall-warn-seconds <n> # Stall warning threshold (default: 30)
--progress-mode <mode>   # Progress output: auto|line|newline (default: auto)
--debug-events           # Log pause/resume/stall events
--quiet                  # Disable progress bars

转换函数

转换函数可让你在导入过程中修改文档。创建一个导出转换函数的 JavaScript 文件：

基本转换（transform.js）

// ES modules (default)
export default function transform(doc) {
  return {
    ...doc,
    full_name: `${doc.first_name} ${doc.last_name}`,
    timestamp: new Date().toISOString(),
  };
}

// Or CommonJS
module.exports = function transform(doc) {
  return {
    ...doc,
    full_name: `${doc.first_name} ${doc.last_name}`,
  };
};

跳过文档

返回 null 或 undefined 以跳过文档：

export default function transform(doc) {
  // Skip invalid documents
  if (!doc.email || !doc.email.includes("@")) {
    return null;
  }
  return doc;
}

拆分文档

返回一个数组，以便从一个源文档创建多个目标文档：

export default function transform(doc) {
  // Split a tweet into multiple hashtag documents
  const hashtags = doc.text.match(/#\w+/g) || [];
  return hashtags.map((tag) => ({
    hashtag: tag,
    tweet_id: doc.id,
    created_at: doc.created_at,
  }));
}

映射

自定义映射（mappings.json）

{
  "properties": {
    "@timestamp": { "type": "date" },
    "message": { "type": "text" },
    "user": {
      "properties": {
        "name": { "type": "keyword" },
        "email": { "type": "keyword" }
      }
    }
  }
}

node scripts/ingest.js ingest --file /absolute/path/to/data.json --target my-index --mappings mappings.json

边界

• 绝不回显、打印、记录或以其他方式泄露凭据环境变量的值 （$ELASTICSEARCH_API_KEY、$ELASTICSEARCH_PASSWORD、$ELASTICSEARCH_CLOUD_ID 等）。不要运行会暴露机密值的 Shell 命令 （例如 echo $ELASTICSEARCH_API_KEY、env | grep KEY、printenv）。导出这些变量并运行在内部读取它们的脚本是预期且安全的——限制仅针对将机密值显示在命令输出中。验证连接的唯一方式是运行 node scripts/ingest.js test。如果测试 失败，请用户检查其环境配置——不要自行尝试诊断凭据问题。
• 未经用户明确确认，绝不运行破坏性命令（例如使用 --delete-index 标志或删除现有索引和数据）。

指南

• 先测试：摄取数据前，始终运行 node scripts/ingest.js test。如果连接失败，请用户检查其环境配置并重新测试。测试通过前不要尝试摄取数据。
• 绝不要同时使用 --infer-mappings 和 --source-format。推断会创建服务端摄取管道来处理解析（例如 CSV 处理器）。使用 --source-format csv 也会在客户端解析，导致重复解析并产生空索引。自动检测时单独使用 --infer-mappings；手动控制时，则将 --source-format 与显式的 --mappings 搭配使用。
• 需要使用已知字段类型在客户端解析 CSV 时，使用 --source-format csv 和 --mappings。
• 需要 Elasticsearch 检测格式、推断字段类型并自动创建摄取管道时，单独使用 --infer-mappings。

不适用场景

以下情况请考虑使用其他方案：

• 重新索引或索引迁移：使用 elasticsearch-reindex 技能来复制、迁移或转换现有 Elasticsearch 索引
• 实时摄取：使用 Filebeat 或 Elastic Agent
• 企业级管道：使用 Logstash
• 内置转换：使用 Elasticsearch Transforms

更多资源

• 常见模式 - CSV 加载、批量摄取、数据丰富等详细示例
• 故障排查 - 常见问题的解决方案

参考资料

• Elasticsearch 映射
• Elasticsearch 查询 DSL