elasticsearch-file-ingest

어떤 스킬인가요?

Ingest and transform data files (CSV/JSON/Parquet/Arrow IPC) into Elasticsearch with stream processing and custom transforms. Use when loading files or batch importing data — not for reindexing, general ingest pipeline design, or bulk API patterns.

설치 방법

프로젝트 폴더에서 아래 명령어를 실행하고, 설치할 에이전트를 선택하세요.

설치 명령을 직접 실행해야 적용됩니다. 지원 에이전트와 필요한 권한·라이선스는 제작자의 안내를 확인하세요.

스킬 원문 파일

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

/**
 * Example transform that conditionally skips documents.
 *
 * This validates documents and only indexes valid ones.
 *
 * Usage:
 *   node scripts/ingest.js ingest --file data.json --target validated --transform examples/skip-transform.js
 */

export default function transform(doc) {
  // Skip documents without required fields
  if (!doc.email || !doc.name) {
    console.warn(`Skipping document without email or name:`, doc.id);
    return null;
  }

  // Skip invalid email addresses
  if (!doc.email.includes("@")) {
    console.warn(`Skipping document with invalid email:`, doc.email);
    return null;
  }

  // Skip test data
  if (doc.email.endsWith("@test.com") || doc.email.endsWith("@example.com")) {
    return null;
  }

  // Return the document if all validations pass
  return {
    ...doc,
    validated_at: new Date().toISOString(),
  };
}

/**
 * Example transform that splits one document into multiple documents.
 *
 * This example takes a tweet and creates a separate document for each hashtag.
 *
 * Usage:
 *   node scripts/ingest.js ingest --file tweets.json --target hashtags --transform examples/split-transform.js
 */

export default function transform(doc) {
  // Extract hashtags from tweet text
  const hashtags = (doc.text || "").match(/#\w+/g) || [];

  // If no hashtags, skip this document
  if (hashtags.length === 0) {
    return null;
  }

  // Create one document per hashtag
  return hashtags.map((tag) => ({
    hashtag: tag.toLowerCase(),
    tweet_id: doc.id,
    user_id: doc.user_id,
    created_at: doc.created_at,
    original_text: doc.text,
  }));
}

/**
 * Example transform function that enriches documents during ingestion.
 *
 * Usage:
 *   node scripts/ingest.js ingest --file data.json --target my-index --transform examples/transform.js
 */

export default function transform(doc) {
  // Add processing metadata
  const enriched = {
    ...doc,
    processed_at: new Date().toISOString(),
    source: "batch-import",
  };

  // Combine first and last name if present
  if (doc.first_name && doc.last_name) {
    enriched.full_name = `${doc.first_name} ${doc.last_name}`;
  }

  // Extract year from timestamp if present
  if (doc.timestamp || doc["@timestamp"]) {
    const timestamp = doc.timestamp || doc["@timestamp"];
    enriched.year = new Date(timestamp).getFullYear();
  }

  // Normalize email to lowercase
  if (doc.email) {
    enriched.email = doc.email.toLowerCase();
  }

  return enriched;
}

// For CommonJS compatibility
// module.exports = transform;

{
  "name": "elasticsearch-file-ingest",
  "version": "0.0.1",
  "description": "Agent skill for ingesting and transforming large data files (CSV/JSON/Parquet/Arrow IPC) into Elasticsearch indices. Stream-based ingestion and custom transformations.",
  "type": "module",
  "private": true,
  "dependencies": {
    "@elastic/elasticsearch": "^8.17.0",
    "node-es-transformer": "^1.2.2"
  }
}

# Common Ingestion Patterns

Detailed examples for common data ingestion scenarios.

## Pattern 1: Load CSV with Custom Mappings

```bash
# 1. Create mappings.json with your schema
cat > mappings.json << 'EOF'
{
  "properties": {
    "timestamp": { "type": "date" },
    "user_id": { "type": "keyword" },
    "action": { "type": "keyword" },
    "value": { "type": "double" }
  }
}
EOF

# 2. Ingest CSV (skip header row)
node scripts/ingest.js ingest \
  --file events.csv \
  --target events \
  --mappings mappings.json \
  --skip-header
```

## Pattern 2: Batch Ingest Multiple Files

```bash
# Ingest all JSON files in a directory
node scripts/ingest.js ingest \
  --file "logs/*.json" \
  --target combined-logs \
  --mappings mappings.json
```

## Pattern 3: Document Enrichment During Ingestion

```bash
# 1. Create enrichment transform
cat > enrich.js << 'EOF'
export default function transform(doc) {
  return {
    ...doc,
    enriched_at: new Date().toISOString(),
    source: 'batch-import',
    year: new Date(doc.timestamp).getFullYear(),
  };
}
EOF

# 2. Ingest with enrichment
node scripts/ingest.js ingest \
  --file data.json \
  --target enriched-data \
  --transform enrich.js
```

## Pattern 4: Performance Tuning

### For Large Files (>5GB)

```bash
# Increase buffer size for better throughput
node scripts/ingest.js ingest \
  --file huge-file.json \
  --target my-index \
  --buffer-size 10240  # 10 MB buffer
```

### Quiet Mode (for scripts)

```bash
# Disable progress bars for automated scripts
node scripts/ingest.js ingest \
  --file data.json \
  --target my-index \
  --quiet
```

# Troubleshooting

Common issues and solutions for the ingest tool.

## Connection Refused

Elasticsearch is not running or the URL is incorrect. Run the connection test:

```bash
node scripts/ingest.js test
```

If the test fails, ask the user to verify their Elasticsearch environment configuration.

## Out of Memory Errors

Reduce buffer size:

```bash
node scripts/ingest.js ingest --file data.json --target my-index --buffer-size 2048
```

## Transform Function Not Loading

Ensure the transform file exports correctly:

```javascript
// ✓ Correct (ES modules)
export default function transform(doc) {
  /* ... */
}

// ✓ Correct (CommonJS)
module.exports = function transform(doc) {
  /* ... */
};

// ✗ Wrong
function transform(doc) {
  /* ... */
}
```

## Mapping Conflicts

Delete and recreate the index:

```bash
node scripts/ingest.js ingest \
  --file data.json \
  --target my-index \
  --mappings mappings.json \
  --delete-index
```

## Slow Ingestion

Check these common causes:

1. **Large documents**: Reduce `--buffer-size`
2. **Complex transforms**: Simplify transform logic
3. **Elasticsearch load**: Check cluster health and indexing queue

## Stall Warnings

If you see stall warnings, the ingestion is pausing due to backpressure:

```bash
# Increase stall warning threshold
node scripts/ingest.js ingest \
  --file data.json \
  --target my-index \
  --stall-warn-seconds 60

# Debug pause/resume events
node scripts/ingest.js ingest \
  --file data.json \
  --target my-index \
  --debug-events
```

## CSV Parsing Issues

For CSV files with non-standard formatting:

```bash
# Create csv-options.json
cat > csv-options.json << 'EOF'
{
  "columns": true,
  "delimiter": ";",
  "trim": true,
  "skip_empty_lines": true
}
EOF

node scripts/ingest.js ingest \
  --file data.csv \
  --source-format csv \
  --csv-options csv-options.json \
  --target my-index
```

## Authentication Errors

Run the built-in connection test to verify credentials and connectivity:

```bash
node scripts/ingest.js test
```

If the test fails, ask the user to verify their Elasticsearch credentials and environment configuration.

#!/usr/bin/env node

import fs from "node:fs";
import path from "node:path";
import transformer from "node-es-transformer";
import { Client } from "@elastic/elasticsearch";

const args = process.argv.slice(2);

function showUsage() {
  console.log("Usage: ingest.js <command> [options]");
  console.log("\nCommands:");
  console.log("  test                     Test Elasticsearch connection");
  console.log("  ingest [options]         Ingest data into Elasticsearch");
  console.log("  help                     Show this help message");
  console.log("\nRequired (ingest):");
  console.log("  --target <index>         Target index name");
  console.log("\nSource (choose one):");
  console.log("  --file <path>            Source file (supports wildcards, e.g., logs/*.json)");
  console.log("  --stdin                  Read NDJSON/CSV from stdin");
  console.log("\nElasticsearch Connection (environment variables only):");
  console.log("  ELASTICSEARCH_API_KEY, ELASTICSEARCH_USERNAME, ELASTICSEARCH_PASSWORD");
  console.log("  ELASTICSEARCH_CLOUD_ID, ELASTICSEARCH_URL, ELASTICSEARCH_INSECURE");
  console.log("\nIndex Configuration:");
  console.log("  --mappings <file.json>   Mappings file");
  console.log("  --infer-mappings         Infer mappings/pipeline from file/stream");
  console.log("  --infer-mappings-options <file>  Options for inference (JSON file)");
  console.log("  --delete-index           Delete target index if exists");
  console.log("  --pipeline <name>        Ingest pipeline name");
  console.log("\nProcessing:");
  console.log("  --transform <file.js>    Transform function (export as default or module.exports)");
  console.log("  --source-format <fmt>    Source format: ndjson|csv|parquet|arrow (default: ndjson)");
  console.log("  --csv-options <file>     CSV parser options (JSON file)");
  console.log("  --skip-header            Skip first line (e.g., CSV header)");
  console.log("\nPerformance:");
  console.log("  --buffer-size <kb>       Buffer size in KB (default: 5120)");
  console.log("  --total-docs <n>         Total docs for progress bar (file/stream)");
  console.log("  --stall-warn-seconds <n> Stall warning threshold (default: 30)");
  console.log("  --progress-mode <mode>   Progress output: auto|line|newline (default: auto)");
  console.log("  --debug-events           Log pause/resume/stall events");
  console.log("  --quiet                  Disable progress bars");
  console.log("\nExamples:");
  console.log("  # Test connection");
  console.log("  ingest.js test");
  console.log("");
  console.log("  # Ingest a JSON file");
  console.log("  ingest.js ingest --file data.json --target my-index");
  console.log("");
  console.log("  # Ingest with custom mappings");
  console.log("  ingest.js ingest --file data.json --target my-index --mappings mappings.json");
  console.log("");
  console.log("  # Ingest with transformation");
  console.log("  ingest.js ingest --file data.json --target my-index --transform transform.js");
  process.exit(1);
}

function getDefaultClientConfig() {
  const cloudId = process.env.ELASTICSEARCH_CLOUD_ID;
  const apiKey = process.env.ELASTICSEARCH_API_KEY;
  const url = process.env.ELASTICSEARCH_URL;
  const username = process.env.ELASTICSEARCH_USERNAME;
  const password = process.env.ELASTICSEARCH_PASSWORD;
  const insecure = process.env.ELASTICSEARCH_INSECURE === "true";

  const config = {};

  if (cloudId) {
    config.cloud = { id: cloudId };
  } else if (url) {
    config.node = url;
  } else {
    config.node = "http://localhost:9200";
  }

  if (apiKey) {
    config.auth = { apiKey };
  } else if (username && password) {
    config.auth = { username, password };
  }

  if (insecure) {
    config.tls = { rejectUnauthorized: false };
  }

  config.headers = { "User-Agent": "elastic-agentic" };

  return config;
}

function parseArgs(args) {
  const options = {
    sourceClientConfig: getDefaultClientConfig(),
    targetClientConfig: null,
    verbose: true,
  };

  for (let i = 0; i < args.length; i++) {
    const arg = args[i];
    const next = args[i + 1];

    switch (arg) {
      case "--file":
        if (!next) showUsage();
        options.fileName = next;
        i++;
        break;

      case "--stdin":
        options.stream = process.stdin;
        break;

      case "--target":
        if (!next) showUsage();
        options.targetIndexName = next;
        i++;
        break;

      case "--mappings":
        if (!next) showUsage();
        try {
          const content = fs.readFileSync(next, "utf8");
          options.mappings = JSON.parse(content);
        } catch (err) {
          console.error(`Error reading mappings file ${next}:`, err.message);
          process.exit(1);
        }
        i++;
        break;

      case "--infer-mappings":
        options.inferMappings = true;
        break;

      case "--infer-mappings-options":
        if (!next) showUsage();
        try {
          const content = fs.readFileSync(next, "utf8");
          options.inferMappingsOptions = JSON.parse(content);
        } catch (err) {
          console.error(`Error reading infer mappings options file ${next}:`, err.message);
          process.exit(1);
        }
        i++;
        break;

      case "--delete-index":
        options.deleteIndex = true;
        break;

      case "--pipeline":
        if (!next) showUsage();
        options.pipeline = next;
        i++;
        break;

      case "--transform":
        if (!next) showUsage();
        try {
          const transformPath = path.resolve(process.cwd(), next);
          // Dynamic import for ES modules
          import(transformPath)
            .then((mod) => {
              options.transform = mod.default || mod;
            })
            .catch((err) => {
              // Fallback to require for CommonJS
              try {
                options.transform = require(transformPath);
              } catch (requireErr) {
                console.error(`Error loading transform file ${next}:`, err.message);
                process.exit(1);
              }
            });
        } catch (err) {
          console.error(`Error loading transform file ${next}:`, err.message);
          process.exit(1);
        }
        i++;
        break;

      case "--source-format":
        if (!next) showUsage();
        options.sourceFormat = next;
        i++;
        break;

      case "--csv-options":
        if (!next) showUsage();
        try {
          const content = fs.readFileSync(next, "utf8");
          options.csvOptions = JSON.parse(content);
        } catch (err) {
          console.error(`Error reading CSV options file ${next}:`, err.message);
          process.exit(1);
        }
        i++;
        break;

      case "--skip-header":
        options.skipHeader = true;
        break;

      case "--buffer-size":
        if (!next) showUsage();
        options.bufferSize = parseInt(next, 10);
        i++;
        break;

      case "--total-docs":
        if (!next) showUsage();
        options.totalDocs = parseInt(next, 10);
        i++;
        break;

      case "--stall-warn-seconds":
        if (!next) showUsage();
        options.stallWarnSeconds = parseInt(next, 10);
        i++;
        break;

      case "--progress-mode":
        if (!next) showUsage();
        options.progressMode = next;
        i++;
        break;

      case "--debug-events":
        options.debugEvents = true;
        break;

      case "--quiet":
        options.verbose = false;
        break;

      case "--help":
      case "-h":
        showUsage();
        break;

      default:
        console.error(`Unknown option: ${arg}\n`);
        showUsage();
    }
  }

  // Auto-detect source format from file extension when not explicitly set
  if (!options.sourceFormat && options.fileName) {
    const ext = path.extname(options.fileName).toLowerCase();
    const formatMap = {
      ".csv": "csv",
      ".json": "ndjson",
      ".ndjson": "ndjson",
      ".parquet": "parquet",
      ".arrow": "arrow",
    };
    if (formatMap[ext]) {
      options.sourceFormat = formatMap[ext];
    }
  }

  // Validation
  if (!options.targetIndexName) {
    console.error("Error: --target is required\n");
    showUsage();
  }

  if (!options.fileName && !options.stream) {
    console.error("Error: Either --file or --stdin is required\n");
    showUsage();
  }

  if (options.fileName && options.stream) {
    console.error("Error: Only one of --file or --stdin can be used\n");
    showUsage();
  }

  return options;
}

async function testConnection(clientConfig) {
  const client = new Client(clientConfig);
  try {
    const info = await client.info();
    return {
      success: true,
      cluster: info.cluster_name,
      version: info.version.number,
      node: clientConfig.node || clientConfig.cloud?.id || "cloud",
    };
  } catch (error) {
    return {
      success: false,
      error: error.message,
      node: clientConfig.node || clientConfig.cloud?.id || "cloud",
    };
  }
}

function printConnectionHelp() {
  console.error("");
  console.error("Set one of these environment variable combinations:");
  console.error("  1. Elastic Cloud: ELASTICSEARCH_CLOUD_ID + ELASTICSEARCH_API_KEY");
  console.error("  2. Direct URL + API Key: ELASTICSEARCH_URL + ELASTICSEARCH_API_KEY");
  console.error("  3. Basic Auth: ELASTICSEARCH_URL + ELASTICSEARCH_USERNAME + ELASTICSEARCH_PASSWORD");
  console.error("");
  console.error("For self-signed certs: set ELASTICSEARCH_INSECURE=true");
  console.error("");
  console.error("For local development, see:");
  console.error("  https://www.elastic.co/guide/en/elasticsearch/reference/current/run-elasticsearch-locally.html");
  console.error("");
  console.error("Then re-run: node scripts/ingest.js test");
}

async function runTest(clientConfig) {
  console.log("=== Testing Elasticsearch Connection ===\n");
  const connTest = await testConnection(clientConfig);

  if (!connTest.success) {
    console.error(`✗ Connection failed to ${connTest.node}`);
    console.error(`  Error: ${connTest.error}`);
    printConnectionHelp();
    process.exit(1);
  }

  console.log("✓ Connected successfully!");
  console.log(`  Cluster: ${connTest.cluster}`);
  console.log(`  Version: ${connTest.version}`);
  console.log(`  Node:    ${connTest.node}`);

  // Test bulk indexing capability with a dummy check
  const client = new Client(clientConfig);
  try {
    const health = await client.cluster.health();
    console.log(`  Status:  ${health.status}`);
    console.log(`  Nodes:   ${health.number_of_nodes}`);
  } catch {
    // cluster.health may fail with limited permissions — not critical
  } finally {
    await client.close();
  }

  console.log("\n✓ Ready for ingestion");
}

async function main() {
  if (args.length === 0 || args.includes("--help") || args.includes("-h") || args[0] === "help") {
    showUsage();
  }

  // Handle "test" subcommand before parsing ingest options
  if (args[0] === "test") {
    await runTest(getDefaultClientConfig());
    return;
  }

  // Require "ingest" subcommand
  if (args[0] !== "ingest") {
    console.error(`Unknown command: ${args[0]}\n`);
    showUsage();
  }

  const options = parseArgs(args.slice(1));

  // Test connection before starting ingestion
  console.log("Testing Elasticsearch connection...");
  const connTest = await testConnection(options.sourceClientConfig);

  if (!connTest.success) {
    console.error(`\n✗ Connection failed to ${connTest.node}`);
    console.error(`  Error: ${connTest.error}`);
    printConnectionHelp();
    process.exit(1);
  }

  console.log(`✓ Connected to ${connTest.cluster} (ES ${connTest.version})\n`);

  try {
    console.log("Starting ingestion...");
    console.log(`Target index: ${options.targetIndexName}`);

    if (options.fileName) {
      console.log(`Source: File ${options.fileName}`);
    } else {
      console.log(`Source: stdin`);
    }

    const result = await transformer(options);

    const enableProgress = options.verbose !== false && Boolean(options.fileName || options.stream);
    const envTotal = Number.parseInt(process.env.ES_TRANSFORMER_TOTAL_DOCS || "", 10);
    const totalDocs = Number.isFinite(options.totalDocs)
      ? options.totalDocs
      : Number.isFinite(envTotal)
        ? envTotal
        : null;

    let processed = 0;
    let lastRate = 0;
    let paused = false;
    let pauseStartedAt = null;
    let lastProgressAt = Date.now();
    let stallLogged = false;
    const startTime = Date.now();
    const debugEvents = options.debugEvents || process.env.ES_TRANSFORMER_DEBUG_EVENTS === "1";
    const progressMode = options.progressMode || process.env.ES_TRANSFORMER_PROGRESS_MODE || "auto";
    const stallWarnSeconds = Number.isFinite(options.stallWarnSeconds)
      ? options.stallWarnSeconds
      : Number.parseInt(process.env.ES_TRANSFORMER_STALL_WARN_SECONDS || "30", 10);

    function formatNumber(value) {
      return new Intl.NumberFormat("en-US").format(value);
    }

    const progressStream = process.stdout.isTTY ? process.stdout : process.stderr;
    const autoLineMode =
      progressMode === "auto" && (progressStream.isTTY || (process.env.TERM && process.env.TERM !== "dumb"));
    const lineMode = progressMode === "line" || autoLineMode;
    let lastLineLength = 0;

    function writeProgressLine(line) {
      if (lineMode) {
        if (progressStream.isTTY) {
          progressStream.clearLine(0);
          progressStream.cursorTo(0);
          progressStream.write(line);
        } else {
          const pad = Math.max(0, lastLineLength - line.length);
          progressStream.write(`\r${line}${" ".repeat(pad)}`);
        }
        lastLineLength = Math.max(lastLineLength, line.length);
        return;
      }
      progressStream.write(`${line}\n`);
    }

    function renderProgress(final = false) {
      const elapsedSeconds = Math.max((Date.now() - startTime) / 1000, 1);
      const avgRate = processed / elapsedSeconds;
      const processedStr = formatNumber(processed);
      const totalStr = totalDocs ? formatNumber(totalDocs) : null;
      const pct = totalDocs && totalDocs > 0 ? Math.min(processed / totalDocs, 1) * 100 : null;
      const statusStr =
        paused && pauseStartedAt ? ` | paused ${Math.round((Date.now() - pauseStartedAt) / 1000)}s` : "";

      const columns = progressStream.isTTY ? progressStream.columns : null;

      let rateStr = `${lastRate.toFixed(1)} docs/s`;
      let avgStr = `avg ${avgRate.toFixed(1)} docs/s`;
      let barWidth = 30;
      let includeAvg = true;
      let includeBar = Boolean(totalDocs && totalDocs > 0);

      function buildLine() {
        if (includeBar && pct !== null) {
          const filled = Math.round((pct / 100) * barWidth);
          const bar = `${"#".repeat(filled)}${" ".repeat(barWidth - filled)}`;
          const base = `[${bar}] ${processedStr}/${totalStr} (${pct.toFixed(1)}%)`;
          const rates = includeAvg ? ` | ${rateStr} (${avgStr})` : ` | ${rateStr}`;
          return `${base}${rates}${statusStr}`;
        }
        const rates = includeAvg ? ` | ${rateStr} (${avgStr})` : ` | ${rateStr}`;
        return `${processedStr} docs${rates}${statusStr}`;
      }

      let line = buildLine();

      if (columns && line.length > columns) {
        while (includeBar && barWidth > 10 && line.length > columns) {
          barWidth -= 5;
          line = buildLine();
        }
      }

      if (columns && line.length > columns && includeAvg) {
        includeAvg = false;
        line = buildLine();
      }

      if (columns && line.length > columns) {
        rateStr = `${lastRate.toFixed(0)}/s`;
        avgStr = `avg ${avgRate.toFixed(0)}/s`;
        line = buildLine();
      }

      if (columns && line.length > columns && includeBar) {
        includeBar = false;
        line = buildLine();
      }

      if (columns && line.length > columns && pct !== null) {
        line = `${processedStr}/${totalStr} ${pct.toFixed(1)}%${statusStr}`;
      }

      writeProgressLine(line);

      if (final && lineMode) {
        progressStream.write("\n");
      }
    }

    let stallTimer = null;

    if (enableProgress) {
      result.events.on("docsPerSecond", (dps) => {
        processed += dps;
        lastRate = dps;
        if (dps > 0) {
          lastProgressAt = Date.now();
          stallLogged = false;
        }
        renderProgress();
      });

      result.events.on("pause", () => {
        paused = true;
        pauseStartedAt = Date.now();
        if (debugEvents) {
          progressStream.write(`\n[event] pause at ${new Date().toISOString()}\n`);
        }
        renderProgress();
      });

      result.events.on("resume", () => {
        paused = false;
        pauseStartedAt = null;
        if (debugEvents) {
          progressStream.write(`\n[event] resume at ${new Date().toISOString()}\n`);
        }
        renderProgress();
      });

      stallTimer = setInterval(() => {
        if (!enableProgress) return;
        if (paused) return;
        const since = (Date.now() - lastProgressAt) / 1000;
        if (since >= stallWarnSeconds && !stallLogged) {
          stallLogged = true;
          const msg = `\n⚠️  No docs indexed for ${Math.round(since)}s. Check ES cluster health or bulk errors.\n`;
          progressStream.write(msg);
          if (debugEvents) {
            progressStream.write(
              `[event] stall detected at ${new Date().toISOString()} (since ${Math.round(since)}s)\n`,
            );
          }
        }
      }, 1000);
    }

    result.events.on("finish", () => {
      if (stallTimer) clearInterval(stallTimer);
      if (enableProgress) {
        renderProgress(true);
      }
      if (debugEvents) {
        progressStream.write(`[event] finish at ${new Date().toISOString()}\n`);
      }
      console.log("✓ Ingestion complete!");
    });
  } catch (err) {
    console.error("✗ Error:", err.message);
    process.exit(1);
  }
}

main();

---
name: elasticsearch-file-ingest
description: >
  Ingest and transform data files (CSV/JSON/Parquet/Arrow IPC) into Elasticsearch
  with stream processing and custom transforms. Use when loading files or batch importing
  data — not for reindexing, general ingest pipeline design, or bulk API patterns.
metadata:
  author: elastic
  version: 0.2.0
---

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

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

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

#### Option 2: Direct URL with API Key

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

#### Option 3: Basic Authentication

```bash
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](https://www.elastic.co/guide/en/elasticsearch/reference/current/run-elasticsearch-locally.html)
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)

```bash
export ELASTICSEARCH_INSECURE="true"
```

## Test Connection

Verify the Elasticsearch connection before ingesting data:

```bash
node scripts/ingest.js test
```

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

## Examples

### Ingest a JSON file

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

### Stream NDJSON/CSV via stdin

```bash
# 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

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

### Ingest Parquet directly

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

### Ingest Arrow IPC directly

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

### Ingest CSV with parser options

```bash
# 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:

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

### Infer mappings with options

```bash
# 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

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

### Ingest with transformation

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

## Command Reference

### Required Options

```bash
--target <index>         # Target index name
```

### Source Options (choose one)

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

### Index Configuration

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

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

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

```javascript
// 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:

```javascript
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:

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

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

```bash
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](https://www.elastic.co/beats/filebeat) or
  [Elastic Agent](https://www.elastic.co/guide/en/fleet/current/fleet-overview.html)
- **Enterprise pipelines**: Use [Logstash](https://www.elastic.co/products/logstash)
- **Built-in transforms**: Use
  [Elasticsearch Transforms](https://www.elastic.co/guide/en/elasticsearch/reference/current/transforms.html)

## Additional Resources

- [Common Patterns](references/patterns.md) - Detailed examples for CSV loading, batch ingestion, enrichment, and more
- [Troubleshooting](references/troubleshooting.md) - Solutions for common issues

## References

- [Elasticsearch Mappings](https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping.html)
- [Elasticsearch Query DSL](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html)