elasticsearch-file-ingest（elasticsearch 文件摄取）

Elasticsearch File Ingest

Stream-based ingestion and transformation of large data files (NDJSON, CSV, Parquet, Arrow IPC) into Elasticsearch.

Features & Use Cases

• Stream-based: Handle large files without running out of memory
• High throughput: 50k+ documents/second on commodity hardware
• Formats: NDJSON, CSV, Parquet, Arrow IPC
• Transformations: Apply custom JavaScript transforms during ingestion (enrich, split, filter)
• Batch processing: Ingest multiple files matching a pattern (e.g., logs/*.json)
• Document splitting: Transform one source document into multiple targets

Prerequisites

• Elasticsearch 8.x or 9.x accessible (local or remote)
• Node.js 22+ installed

Setup

This skill is self-contained. The scripts/ folder and package.json live in this skill's directory. Run all commands from this directory. Use absolute paths when referencing data files located elsewhere.

Before first use, install dependencies:

npm install

Environment Configuration

Elasticsearch connection is configured by users exclusively via environment variables. Never pass credentials as command-line arguments. If the test fails, output the setup options below to the user, then stop. Do not proceed with ingestion until a successful connection test.

Option 1: Elastic Cloud (recommended for production)

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

Option 2: Direct URL with API Key

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

Option 3: Basic Authentication

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

Option 4: Local Development

For local development and testing, see Run Elasticsearch locally to spin up Elasticsearch and Kibana. After setup, export the connection variables (URL and API key or credentials) as shown in Option 2 or Option 3 above.

Optional: Skip TLS verification (development only)

export ELASTICSEARCH_INSECURE="true"

Test Connection

Verify the Elasticsearch connection before ingesting data:

node scripts/ingest.js test

Always run this first. If the test fails, resolve the connection issue before proceeding.

Examples

Ingest a JSON file

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

Stream NDJSON/CSV via stdin

# 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

Ingest CSV directly

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

Ingest Parquet directly

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

Ingest Arrow IPC directly

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

Ingest CSV with parser options

# 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

Infer mappings/pipeline from CSV

When using --infer-mappings, do not combine it with --source-format csv. Inference sends a raw sample to Elasticsearch's _text_structure/find_structure endpoint, which returns both mappings and an ingest pipeline with a CSV processor. If --source-format csv is also set, CSV is parsed client-side and server-side, resulting in an empty index. Let --infer-mappings handle everything:

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

Infer mappings with options

# 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

Ingest with custom mappings

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

Ingest with transformation

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

Command Reference

Required Options

--target <index>         # Target index name

Source Options (choose one)

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

Index Configuration

--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

Processing

--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)

Performance

--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

Transform Functions

Transform functions let you modify documents during ingestion. Create a JavaScript file that exports a transform function:

Basic Transform (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}`,
  };
};

Skip Documents

Return null or undefined to skip a document:

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

Split Documents

Return an array to create multiple target documents from one source:

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

Custom Mappings (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

Boundaries

• Never echo, print, log, or otherwise reveal the values of credential environment variables ($ELASTICSEARCH_API_KEY, $ELASTICSEARCH_PASSWORD, $ELASTICSEARCH_CLOUD_ID, etc.). Do not run shell commands whose output would expose secret values (e.g., echo $ELASTICSEARCH_API_KEY, env | grep KEY, printenv). Exporting these variables and running scripts that read them internally is expected and safe — the restriction is on surfacing secret values in command output. The only way to verify connectivity is node scripts/ingest.js test. If the test fails, ask the user to check their environment configuration — do not attempt to diagnose credentials yourself.
• Never run destructive commands (such as using the --delete-index flag or deleting existing indices and data) without explicit user confirmation.

Guidelines

• Test first: Always run node scripts/ingest.js test before ingesting data. If the connection fails, ask the user to verify their environment configuration and re-test. Do not attempt ingestion until the test passes.
• Never combine --infer-mappings with --source-format. Inference creates a server-side ingest pipeline that handles parsing (e.g., CSV processor). Using --source-format csv parses client-side as well, causing double-parsing and an empty index. Use --infer-mappings alone for automatic detection, or --source-format with explicit --mappings for manual control.
• Use --source-format csv with --mappings when you want client-side CSV parsing with known field types.
• Use --infer-mappings alone when you want Elasticsearch to detect the format, infer field types, and create an ingest pipeline automatically.

When NOT to Use

Consider alternatives for:

• Reindexing or index migration: Use the elasticsearch-reindex skill for copying, migrating, or transforming existing Elasticsearch indices
• Real-time ingestion: Use Filebeat or Elastic Agent
• Enterprise pipelines: Use Logstash
• Built-in transforms: Use Elasticsearch Transforms

Additional Resources

• Common Patterns - Detailed examples for CSV loading, batch ingestion, enrichment, and more
• Troubleshooting - Solutions for common issues

References

• Elasticsearch Mappings
• Elasticsearch Query DSL