elasticsearch-ingest

Elasticsearch ファイル取り込み

ローカルのデータファイルをバルク NDJSON に変換し、型が重要な場合は適切なマッピングでインデックスを作成してから、ドキュメントを一括インデックスし、結果を確認して Elasticsearch に読み込みます。

環境設定

このスキルは、elastic CLI を使用して Elasticsearch の操作を実行します。 elastic CLI がインストールされていない場合は、何のために必要かをユーザーに伝えてください。認証情報を推測したり、HTTP API を直接呼び出したり、ほかの回避策を試みたりしないでください。

このスキルでは、HTTP の省略形で操作を記載しています（例：GET /、GET /_cat/indices、GET /{index}/_mapping、 GET /{index}/_settings/index.mode、POST /_query）。このドキュメント末尾の操作表に、各省略形に対応する elastic CLI コマンドが記載されています。HTTP API を直接呼び出さず、必ず CLI を使用してください。

対象範囲

このスキルは、POST /_bulk を使用したファイル → インデックスへの読み込みを対象としています。Logstash、Filebeat、Elastic Agent、Node.js の取り込みツール、その他のサイドカーパイプラインは使用しません。既存のインデックス間でドキュメントをコピーする場合は、ソースファイルを再解析するのではなく、インデックス間の reindex を使用してください。

対応するソース形式：

ソース形式	例	Bulk の要件
ヘッダー行付き CSV	id,name,age,... の後にデータ行	ヘッダーをフィールド名として解析し、データ行ごとにアクション行 1 行と JSON オブジェクト 1 つを出力します
JSON 配列ファイル	[{"a":1},{"a":2}]	ドキュメントごとの行に分割します。生の配列を単一のドキュメントとして一括読み込みしないでください
NDJSON / JSON Lines	1 行に JSON オブジェクト 1 つ	アクション行がない場合は追加できます。それ以外は Bulk 用にそのまま使用できます

Parquet、Arrow、その他のバイナリ列指向形式は対象外です。ただし、ユーザーが事前に CSV または JSON に変換する場合は除きます。

プロセス

1. 接続を確認します。 GET / を呼び出します。呼び出しに失敗した場合は、ファイルの読み取りやクラスター状態の変更を行わず、停止して CLI の設定を解決してください。

2. ソースファイルを調べ、形式を分類します。 ファイルを開く（または先頭行を確認する）ことで、次のいずれかを判断します。

   • CSV — 1 行目がカンマ区切りのヘッダーで、それ以降の行がレコードです。データ行数（ヘッダーを除く）を数え、読み込み後にこの件数を報告します。
   • JSON 配列 — ファイルが [ で始まり、オブジェクトの配列が含まれています。配列要素を数えます。各要素はインデックス付きドキュメントになります。
   • NDJSON — 1 行につき 1 つの 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 で明示的な型が必要な場合、またはインデックスが存在しない場合は、バルクロードの前に mappings ブロックを指定して PUT /{index} を呼び出します。CSV の文字列セルから long、date、boolean を動的マッピングで推測させないでください。動的マッピングでは、曖昧な文字列が .keyword サブフィールドを持つ text にマッピングされることがよくあります。

6. ファイルをバルク用 NDJSON に変換します。 各ドキュメントを必ず 2 行で構成する一時的な NDJSON ファイルを書き込みます。

   • 1 行目 — アクションメタデータ。例：{"index":{"_index":"<index>"}}（安定した ID がユーザーに必要な場合に限り、"_id" を追加します）。
   • 2 行目 — 正しく型付けされた値を含むドキュメント JSON（数値は JSON の数値、ブール値は true／false、日付は 2023-01-15 のような ISO-8601 形式の文字列）。

   CSV の場合は、ヘッダー行を JSON フィールド名に対応付け、ステップ 5 のマッピングに合う JSON 型にセルの値を変換します。JSON 配列の場合は、各配列要素を反復処理し、アクション行とオブジェクト行のペアを出力します。NDJSON Bulk Format の実例を参照してください。

7. ドキュメントを一括でインデックス化します。 ステップ 6 で作成した NDJSON ファイルを使って POST /_bulk を呼び出します。応答を確認し、errors が true の場合は、各項目の error オブジェクトを読み取り、マッピングまたはドキュメントの問題を修正してから、修正後に失敗した項目を再試行します。終了コードが 0 だからといって、成功したと判断してはいけません。

8. 結果を検証します。 必ず読み込みを確認してください。ファイルを調べただけで件数を報告してはいけません。

   • GET /{index}/_count を呼び出し、ステップ 2 で想定した行数／要素数と比較します。
   • 型付きの列が重要な場合は、GET /{index}/_mapping を呼び出し、age などのフィールドが数値型（long／integer）、日付が date、ブール値が boolean であり、text ではないことを確認します。

   検証済みのドキュメント件数と、必要に応じて確認済みのフィールド型を報告します。件数またはマッピングの確認に失敗した場合は、Troubleshooting を参照してください。

ガイドライン

• 一括処理のみです。 すべてのファイル読み込みは、NDJSON のアクション行を使って POST /_bulk 経由で行います。バッチファイルに単一ドキュメントの PUT を繰り返す方法や、クライアント側での CSV 解析の代わりに取り込みパイプラインを使う方法、未加工のソースファイルをそのまま送信する方法は使いません。
• JSON 配列は分割する必要があります。 4 要素の配列を 1 ドキュメントとして一括読み込みすると、件数は 1 になります。正しく読み込めば、件数は 4 になります。
• CSV ヘッダーはスキーマです。 最初の CSV 行はフィールド名で、残りの各行が 1 ドキュメントです。ヘッダー 1 行とデータ 5 行のファイルは、取り込み後に件数 5 と報告される必要があります。
• 型変換はドキュメントの JSON 内で行います。 CSV のセルは文字列として取り込まれます。マッピングで long、date、boolean を宣言している場合は、一括リクエスト本文で JSON の数値、ISO 形式の日付文字列、ブール値リテラルを出力してください。動的マッピングで text が選択された後、引用符付き CSV 文字列から Elasticsearch が型を推測することには頼らないでください。
• 型指定された CSV には明示的なマッピングを優先します。 先に PUT /{index} でインデックスを作成すると、範囲クエリや集計を妨げる、すべての値がテキストとしてインデックスされる事態を防げます。
• 冪等性のある再読み込みを行います。 既存のインデックスに再読み込みする場合は、データを削除する前にユーザーに確認してください。_id を指定しない限り、一括リクエストの index アクションは新しいドキュメントを追加します。

例

型指定された列を含む CSV

ソース（users.csv — ヘッダー 1 行＋データ 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 }
]

配列全体に対する 1 組ではなく、配列の 4 要素それぞれに対する 4 組の一括処理用行ペアに変換します。GET /{index}/_count が 4 を返すことを確認します。NDJSON 一括形式を参照してください。

準備済みの NDJSON

ファイルがアクション行とドキュメント行を交互に含む場合は、形式を検証し、対象インデックスとマッピングを確認したうえで、POST /_bulk に直接渡します。

使用しない場合

• 継続的な取り込みやストリーミング取り込み — ログやメトリクスを継続的に収集するには、Elastic Agent または Beats を使用してください。
• 複雑なエンリッチメントパイプライン — サーバー側の取り込みパイプラインは別途設計してください。このスキルでは、読み込み前にファイルをクライアント側で一括処理用 NDJSON に変換します。
• インデックス間のコピーやマッピングの移行 — ファイルにエクスポートするのではなく、インデックス間で再インデックスしてください。
• 非常に大きなバイナリ列形式ファイル — 先にオフラインで CSV または JSON に変換してから、このスキルに従ってください。

参考資料

• NDJSON 一括形式 — CSV と JSON 配列の変換、アクション行の構文、バッチサイズ
• 取り込みのためのマッピング設計 — CSV の型に対する明示的なマッピング、eval 形式のスキーマ
• トラブルシューティング — 件数の誤り、数値が text 型になる問題、一括処理項目のエラー

操作

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