elasticsearch-file-ingest

Elasticsearch File Ingest

大容量データファイル（NDJSON、CSV、Parquet、Arrow IPC）を Elasticsearch にストリーム方式で取り込み、変換します。

特徴とユースケース

• ストリーム方式：メモリ不足を起こさずに大容量ファイルを処理します
• 高スループット：一般的なハードウェアで毎秒 50,000 件以上のドキュメントを処理します
• 形式：NDJSON、CSV、Parquet、Arrow IPC
• 変換：取り込み中にカスタム JavaScript 変換を適用します（拡充、分割、フィルタリング）
• バッチ処理：パターンに一致する複数のファイルを取り込みます（例：logs/*.json）
• ドキュメントの分割：1 つのソースドキュメントを複数のターゲットに変換します

前提条件

• 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

ソースオプション（いずれか 1 つを選択）

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

ドキュメントを分割する

配列を返すと、1 つのソースから複数のターゲットドキュメントを作成できます。

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 など）の値を、絶対にエコー、出力、ログ記録、その他の方法で公開しないでください。秘密の値が出力に表示されるシェルコマンド（例：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 のみを使うか、手動で制御する場合は明示的な --mappings とともに --source-format を使ってください。
• 既知のフィールド型でクライアント側の CSV 解析を行う場合は、--source-format csv を --mappings とともに使ってください。
• Elasticsearch に形式の検出、フィールド型の推論、取り込みパイプラインの自動作成を行わせる場合は、--infer-mappings のみを使ってください。

使用しない方がよい場合

次の場合は、別の方法を検討してください。

• 再インデックス作成やインデックスの移行：既存の Elasticsearch インデックスのコピー、移行、変換には elasticsearch-reindex スキルを使ってください
• リアルタイム取り込み：Filebeat または Elastic Agent を使ってください
• エンタープライズパイプライン：Logstash を使ってください
• 組み込みの変換機能：Elasticsearch Transforms を使ってください

追加リソース

• よくあるパターン - CSV の読み込み、バッチ取り込み、エンリッチメントなどの詳細な例
• トラブルシューティング - よくある問題の解決方法

参考資料

• Elasticsearch マッピング
• Elasticsearch Query DSL