Elastic Certified Engineer Exam対策 – Ingest Pipeline

Training banner: large white text 'Elastic Certified Engineer Exam' with a presenter icon and 'TRAINING' vertical label on right set against a dark tech background. トレーニング

このブログの目的

Elastic Certified Engineer Examでは、Ingest Pipelineに関する問題として、「要件を満たすPipelineを実装する」ことが求められます。Processorを適切に組み合わせてDocumentを加工・変換し、要件どおりのデータをIndexへ登録できることが重要です。本記事では、Elastic社が公開している試験ガイドの以下の試験範囲に関連する内容を扱います。

Data Processing – Define and use an ingest pipeline that satisfies a given set of requirements

本記事では、Ingest Pipelineの基本構成から、代表的なProcessorの利用方法、Pipelineの実装・テスト・デバッグまでを、実際に実行できるサンプルを交えながら解説します。また、試験対策として押さえておきたいポイントだけでなく、実運用でも役立つPipelineの設計やデバッグ方法についても取り上げます。

Ingest Pipelineは、Elastic AgentやBeatsから取り込んだログの前処理だけでなく、アプリケーションから登録されるDocumentの加工にも広く利用されています。Processorごとの役割を理解し、要件に応じて適切に組み合わせられるようになることが、本記事の目的です。

Ingest Pipelineの概要と基本構成

Ingest Pipelineとは

Ingest Pipelineは、DocumentがIndexへ登録される直前に、データの加工や変換を自動的に実行するための機能です。Processorと呼ばれる処理を組み合わせることで、ログの解析、不要なFieldの削除、Field名の変更、データ型の変換、位置情報の付与など、さまざまな前処理をElasticsearch側で実施できます。これにより、アプリケーションやデータ送信側で実装していた前処理をElasticsearch側へ集約でき、一貫したデータ品質を維持できます。

Elastic Certified Engineer Examでは、要件に応じて適切なProcessorを選択し、Pipelineを構築できることが求められます。本記事では、grok、date、set、remove、rename、geoipなどの代表的なProcessorを中心に、その役割や利用方法を解説します。

本章では、まずIngest Pipeline全体の構成と処理の流れを理解し、その後、Pipelineの基本構文や利用方法、代表的なProcessorについて解説します。第3章では、試験で重要となるProcessorを個別に取り上げ、それぞれの使い方やポイントを詳しく説明します。

Ingest Pipelineの処理の流れ

Ingest Pipelineは、DocumentがIndexへ登録される直前に実行される前処理の仕組みです。クライアントから送信されたDocumentは、指定されたPipeline内で定義されたProcessorを順番に実行し、加工・変換された後にMappingが適用され、最終的にIndexへ登録されます。Processorは定義された順序で実行されるため、前のProcessorで生成したFieldを後続のProcessorで利用することも可能です。

一般的な処理の流れは以下のようになります。

  1. Documentの受信 アプリケーションやElastic Agent、BeatsなどからJSON形式のDocumentがElasticsearchへ送信されます。
  2. Pipelineの決定 Index APIで指定されたPipelineや、Index Templateで設定されたdefault_pipeline、final_pipelineに従って適用するPipelineが決定されます。
  3. Processorの順次実行 Pipeline内で定義されたProcessorが上から順に実行され、Fieldの追加・削除、ログの解析、データ型の変換などの前処理が行われます。
  4. Mappingの適用 Pipelineによる加工後のDocumentに対してMappingが適用されます。Mappingに従ってFieldが解釈され、必要に応じてDynamic Mappingが適用されます。
  5. Indexへの登録 すべての処理が正常に完了すると、加工済みのDocumentがIndexへ登録されます。

なお、PipelineはMappingよりも前に実行されるため、Processorで追加・変更したFieldにも、その後Mappingが適用され、定義されたField Typeに従ってIndexへ登録されます。この実行順序を理解しておくことは、適切なPipelineを設計するうえでも、Elastic Certified Engineer Examの対策としても重要です。

Pipelineの基本構文

Ingest Pipelineは、JSON形式で定義し、_ingest/pipeline APIを利用して作成・更新します。Pipelineには複数のProcessorを定義でき、DocumentはProcessorを上から順に実行しながら加工・変換されます。Processorごとに対象Fieldや処理内容を指定することで、Fieldの追加・削除、名称変更、データ型の変換、ログの解析など、さまざまな前処理を柔軟に組み合わせることができます。

以下は、Ingest Pipelineの基本的な構文例です。

PUT _ingest/pipeline/sample-pipeline
{
  "description": "Sample Ingest Pipeline",
  "processors": [
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "rename": {
        "field": "level",
        "target_field": "log_level"
      }
    },
    {
      "remove": {
        "field": "temp_field"
      }
    }
  ]
}

主な構成要素は以下のとおりです。

構成要素説明
Pipeline IDPipelineを識別する一意のIDです。PUT _ingest/pipeline/<pipeline_id>で指定します。
descriptionPipelineの説明です。省略可能ですが、用途を記載しておくと管理しやすくなります。
processors実行するProcessorを配列形式で定義します。Processorは記載した順番に実行されます。
processor実際の前処理を行う要素です。上記例では、set、rename、removeの3つを定義しています。

Processorは定義した順番に実行されるため、前のProcessorで追加・変更したFieldを後続のProcessorで利用できます。そのため、Pipelineを設計する際は、各Processorの依存関係や実行順序を意識することが重要です。

次節では、作成したPipelineをどのように利用するのかについて、Index APIやReindex API、Index Templateによるdefault_pipelineおよびfinal_pipelineの設定方法を解説します。

Pipelineの利用方法

作成したIngest Pipelineは、さまざまな方法で利用できます。最も一般的なのはIndex APIでPipelineを指定する方法ですが、既存データの再加工を目的としてReindex APIから利用することもできます。また、Index Templateにdefault_pipelineやfinal_pipelineを設定することで、Pipelineを自動的に適用することも可能です。

Document登録時にPipelineを指定する

Index APIでは、pipelineパラメータを指定することで、Document登録時にPipelineを適用できます。

POST sample-index/_doc?pipeline=sample-pipeline
{
  "level": "INFO",
  "message": "Application started",
  "temp_field": "temporary data"
}

この例では、Documentが登録される前にsample-pipelineが実行されます。その結果、

  • environmentが追加される
  • levelがlog_levelへ変更される
  • temp_fieldが削除される

という前処理が実行された後、Indexへ登録されます。

Reindex APIで利用する

Reindex APIでは、既存のDocumentを別のIndexへコピーする際にPipelineを適用できます。

POST _reindex
{
  "source": {
    "index": "sample-index"
  },
  "dest": {
    "index": "sample-index-v2",
    "pipeline": "sample-pipeline"
  }
}

この例では、sample-indexのDocumentをsample-index-v2へコピーする際に、sample-pipelineによる前処理が実行されます。

既存データへ新しいFieldを追加したい場合や、Field名を変更したい場合などに利用されます。

Index Templateでdefault_pipelineを指定する

default_pipelineを設定すると、そのIndexへDocumentが登録される際、自動的にPipelineが実行されます。クライアント側でPipelineを指定する必要がなくなるため、前処理を統一したい場合に利用されます。

PUT _index_template/sample-template
{
  "index_patterns": [
    "logs-*"
  ],
  "template": {
    "settings": {
      "index.default_pipeline": "sample-pipeline"
    }
  }
}

この設定後、logs-*に一致するIndexへDocumentを登録すると、自動的にsample-pipelineが適用されます。

final_pipelineを指定する

final_pipelineは、Index APIでpipelineが指定されているか、default_pipelineが設定されているかに関係なく、最後に必ず実行されるPipelineです。

PUT _index_template/sample-template
{
  "index_patterns": [
    "logs-*"
  ],
  "template": {
    "settings": {
      "index.final_pipeline": "sample-pipeline"
    }
  }
}

例えば、

  • クライアントがpipelineを指定した場合
  • default_pipelineが実行された場合

であっても、その後にfinal_pipelineが実行されます。

そのため、監査情報の追加や必須Fieldの設定など、すべてのDocumentに対して必ず実施したい共通処理を定義する用途に適しています。

Elastic Docs > Manage data > Ingest: Bring your data to Elastic > Transform and enrich data > Elasticsearch ingest pipelines

https://www.elastic.co/docs/manage-data/ingest/transform-enrich/ingest-pipelines?utm_source=chatgpt.com#set-final-pipeline

Elastic Docs > Reference > Index settings > General

https://www.elastic.co/docs/reference/elasticsearch/index-settings/index-modules#index-final-pipeline

公式ドキュメントより抜粋

index.default_pipeline

Default ingest pipeline for the index.Index requests will fail if the default pipeline is set and the pipeline does not exist.The default may be overridden using the pipeline parameter.The special pipeline name _none indicates no default ingest pipeline will run.

index.final_pipeline

Final ingest pipeline for the index.Indexing requests will fail if the final pipeline is set and the pipeline does not exist.The final pipeline always runs after the request pipeline (if specified) and the default pipeline (if it exists).The special pipeline name _none indicates no final ingest pipeline will run.

Note

You can’t use a final pipeline to change the _index field. If the pipeline attempts to change the _index field, the indexing request will fail.

index.default_pipeline

インデックスに対するデフォルトのIngest Pipelineです。

“default_pipeline”が設定されているにもかかわらず、そのPipelineが存在しない場合、インデックス登録リクエストは失敗します。

“pipeline”パラメータを使用することで、このデフォルト設定を上書きできます。特別なPipeline名”_none”を指定すると、デフォルトのIngest Pipelineは実行されません。

index.final_pipeline

インデックスに対する最終的なIngest Pipelineです。”final_pipeline”が設定されているにもかかわらず、そのPipelineが存在しない場合、インデックス登録リクエストは失敗します。

“final_pipeline”は、リクエストで指定されたPipeline（指定されている場合）および”default_pipeline”（存在する場合）の実行後に、必ず実行されます。特別なPipeline名”_none”を指定すると、”final_pipeline”は実行されません。

注記

“final_pipeline”を使用して”_index”フィールドを変更することはできません。Pipelineが”_index”フィールドの変更を試みた場合、インデックス登録リクエストは失敗します。

Processor一覧

Ingest Pipelineには、データの加工や変換を行うためのさまざまなProcessorが用意されています。Processorごとに役割が異なり、Fieldの追加・削除やデータ型の変換、ログ解析、位置情報の付与など、用途に応じて組み合わせることで柔軟な前処理を実現できます。

Elastic公式ドキュメントでは、以下のProcessorが提供されています。実運用では用途に応じて利用するProcessorを選択しますが、Elastic Certified Engineer Examでは、これらすべてを暗記する必要はありません。どのようなProcessorが存在するか把握しておく程度で十分です。

試験で出題されるProcessorは、代表的なものが中心になると考えられます。しかし、要件に応じて適切なProcessorを選択することが求められるため、万が一、見慣れないProcessorが出題された場合にも対応できるよう、一通りどのようなProcessorが用意されているか目を通しておくことをお勧めします。すべてを暗記する必要はありません。試験では公式ドキュメントを参照できますので、公式ドキュメントに掲載されているサンプルを参考にしながら、要件に応じたPipelineを実装できるようになれば十分です。

No.Processor説明
1Append配列Fieldへ値を追加する
2Attachment添付ファイルからテキストやメタデータを抽出する
3Bytes容量文字列をバイト数へ変換する
4Circle円形領域をGeo Shapeへ変換する
5CEFCommon Event Format（CEF）ログを解析する
6Community IDネットワークフローのCommunity IDを生成する
7ConvertFieldのデータ型を変換する
8CSVCSV形式の文字列を複数Fieldへ分割する
9Date日時文字列を日時型へ変換する
10Date index name日時からIndex名を生成する
11Dissectパターンに従って文字列を高速に分割する
12Dot expanderドット区切りFieldをオブジェクトへ展開する
13Drop条件に一致したDocumentを破棄する
14EnrichEnrich Policyを利用してデータを付加する
15Fail任意のエラーメッセージでPipelineを失敗させる
16Fingerprint複数Fieldからハッシュ値を生成する
17Foreach配列要素ごとにProcessorを繰り返し実行する
18Geo-gridGeo Grid情報をGeo Shapeへ変換する
19GeoIPIPアドレスから位置情報を付与する
20Grok非構造化ログを解析してFieldへ分割する
21Gsub正規表現を利用して文字列を置換する
22HTML stripHTMLタグを除去する
23Inference学習済みMLモデルによる推論を実行する
24IP LocationIPアドレスから位置情報を取得する
25Join配列を指定した区切り文字で連結する
26JSONJSON文字列をオブジェクトへ変換する
27KVKey-Value形式の文字列を解析する
28Lowercase文字列を小文字へ変換する
29Network directionネットワーク通信方向を判定する
30Normalize for StreamStream向けにDocumentを正規化する
31Pipeline別のIngest Pipelineを呼び出す
32Redact機密情報をマスキングする
33Registered domainFQDNから登録ドメインを抽出する
34Recover Failure DocumentFailure StoreからDocumentを復元する
35Remove不要なFieldを削除する
36RenameField名を変更する
37RerouteDocumentの送信先Data StreamまたはIndexを変更する
38ScriptPainless Scriptによる任意の加工を行う
39SetFieldの追加や値の設定を行う
40Set security user認証ユーザー情報をFieldへ追加する
41Sort配列要素を並び替える
42Split文字列を区切り文字で分割する
43TerminatePipelineの処理を途中で終了する
44Trim文字列の前後の空白を削除する
45Uppercase文字列を大文字へ変換する
46URL decodeURLエンコード文字列をデコードする
47URI partsURIを構成要素へ分割する
48User agentUser-Agent文字列を解析する

次章では、Elastic Certified Engineer Examの学習において優先的に理解しておきたい代表的なProcessorを取り上げ、それぞれの役割や主要なオプション、利用方法についてサンプルを交えながら解説します。

Elastic Docs > Reference > Processor reference

https://www.elastic.co/docs/reference/enrich-processor

試験で押さえるべきProcessor

前章では、Ingest Pipelineの概要や基本構文、利用方法、そしてElasticsearchで利用できるProcessorの一覧について解説しました。本章では、その中でもElastic Certified Engineer Examの学習において優先的に理解しておきたい代表的なProcessorを取り上げ、それぞれの役割や利用方法について解説します。

Elasticには数多くのProcessorが用意されていますが、試験ではすべてのProcessorについて詳細な知識が求められるわけではありません。重要なのは、要件に応じて適切なProcessorを選択し、Ingest Pipelineを実装できることです。そのため、本章では実運用でも利用頻度が高く、試験対策として優先的に理解しておきたいProcessorを中心に取り上げます。

各Processorについては、概要、実行可能なサンプル、主要なオプション、そして試験で押さえておきたいポイントを順に解説します。サンプルを実際に実行しながら動作を確認することで、Processorの役割や使い分けをより理解しやすくなるでしょう。

set Processor

Fieldを追加・更新するProcessor

set Processorは、Fieldの追加や値の設定を行うProcessorです。指定したFieldが存在しない場合は新規に作成され、存在する場合は値が更新されます。固定値の追加や、環境名・システム名などの共通情報をDocumentへ付与する際によく利用されます。構文もシンプルで、Ingest Pipelineの基本的なProcessorの一つです。

サンプルPipeline

PUT _ingest/pipeline/set-pipeline
{
  "description": "Set Processor Sample",
  "processors": [
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=set-pipeline
{
  "message": "Application started"
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/set-processor

rename Processor

Field名を変更するProcessor

rename Processorは、Field名を変更するProcessorです。Documentの内容を変更することなく、Field名のみを別の名称へ変更できます。ログフォーマットの統一や、既存システムとの互換性を保つためによく利用されます。

サンプルPipeline

PUT _ingest/pipeline/rename-pipeline
{
  "description": "Rename Processor Sample",
  "processors": [
    {
      "rename": {
        "field": "level",
        "target_field": "log_level"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=rename-pipeline
{
  "level": "INFO",
  "message": "Application started"
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/rename-processor

remove Processor

不要なFieldを削除するProcessor

remove Processorは、不要なFieldをDocumentから削除するProcessorです。検索や分析に不要な情報を除外したい場合や、保存容量を削減したい場合に利用されます。不要なFieldをIndexへ登録しないことで、データをシンプルに保つことができます。

サンプルPipeline

PUT _ingest/pipeline/remove-pipeline
{
  "description": "Remove Processor Sample",
  "processors": [
    {
      "remove": {
        "field": "temp_field"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=remove-pipeline
{
  "message": "Application started",
  "temp_field": "temporary"
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/remove-processor

convert Processor

Fieldのデータ型を変換するProcessor

convert Processorは、Fieldのデータ型を変換するProcessorです。例えば文字列として受信した数値を整数型へ変換したり、真偽値へ変換したりできます。Mappingへ登録される前にデータ型を整える用途で利用されます。

サンプルPipeline

PUT _ingest/pipeline/convert-pipeline
{
  "description": "Convert Processor Sample",
  "processors": [
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=convert-pipeline
{
  "response_time": "150"
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/convert-processor

date Processor

日時文字列を日時型へ変換するProcessor

date Processorは、日時文字列をElasticsearchの日時型へ変換するProcessorです。複数の日時フォーマットを指定できるため、異なる形式のログを同じ日時Fieldへ統一できます。時系列データを扱う際に利用頻度の高いProcessorです。

サンプルPipeline

PUT _ingest/pipeline/date-pipeline
{
  "description": "Date Processor Sample",
  "processors": [
    {
      "date": {
        "field": "log_time",
        "formats": [
          "yyyy-MM-dd HH:mm:ss"
        ]
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=date-pipeline
{
  "log_time": "2026-06-30 12:30:15"
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/date-processor

grok Processor

非構造化ログを構造化データへ変換するProcessor

grok Processorは、正規表現ベースのGrokパターンを利用して、非構造化ログを複数のFieldへ分割するProcessorです。ApacheやNginxなどのアクセスログや、アプリケーションログを構造化データへ変換する用途で広く利用されています。Ingest Pipelineを代表するProcessorの一つであり、Elastic Certified Engineer Examでも優先的に理解しておきたいProcessorです。

サンプルPipeline

PUT _ingest/pipeline/grok-pipeline
{
  "description": "Grok Processor Sample",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{WORD:method} %{URIPATHPARAM:request}"
        ]
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=grok-pipeline
{
  "message": "192.168.1.10 GET /index.html"
}

実行後は、以下のようなFieldが生成されます。

  • client_ip
  • method
  • request

利用可能なGrokパターンの確認

Grok Processorでは、多数の組み込みGrokパターンがあらかじめ用意されています。利用可能なGrokパターンの一覧は、以下のAPIで確認できます。試験では公式ドキュメントを参照できますが、このAPIを利用することで、実行環境から利用可能なGrokパターンを直接確認することも可能です。

GET _ingest/processor/grok?s

このAPIを実行すると、パターン名でソートされた組み込みGrokパターンの一覧が返されます。見慣れないGrokパターンが必要になった場合でも、一覧を参照しながら適切なパターンを選択できます。特に試験では、すべてのGrokパターンを暗記する必要はなく、このAPIや公式ドキュメントを活用しながら要件に応じたパターンを選択できれば十分です。

Reference

https://www.elastic.co/docs/reference/enrich-processor/grok-processor

3.7 geoip Processor

IPアドレスから位置情報を付与するProcessor

geoip Processorは、IPアドレスから国名や都市名、緯度・経度などの位置情報を取得し、Documentへ付与するProcessorです。Webアクセスログやネットワークログの可視化・分析で広く利用されており、Kibana Mapsとの組み合わせでもよく利用されます。

サンプルPipeline

PUT _ingest/pipeline/geoip-pipeline
{
  "description": "GeoIP Processor Sample",
  "processors": [
    {
      "geoip": {
        "field": "client_ip"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=geoip-pipeline
{
  "client_ip": "202.232.2.100"
}

実行結果 – Mapping

{
  "sample-index": {
    "mappings": {
      "properties": {
        "client_ip": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "geoip": {
          "properties": {
            "city_name": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "continent_name": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "country_iso_code": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "country_name": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "location": {
              "properties": {
                "lat": {
                  "type": "float"
                },
                "lon": {
                  "type": "float"
                }
              }
            },
            "region_iso_code": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "region_name": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            }
          }
        }
      }
    }
  }
}

実行結果 – Data

{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "sample-index",
        "_id": "FwyHF58BsDOBCSPOOWnJ",
        "_score": 1,
        "_source": {
          "geoip": {
            "continent_name": "Asia",
            "region_iso_code": "JP-13",
            "city_name": "Nakano City",
            "country_iso_code": "JP",
            "country_name": "Japan",
            "location": {
              "lat": 35.704,
              "lon": 139.6713
            },
            "region_name": "Tokyo"
          },
          "client_ip": "202.232.2.100"
        }
      }
    ]
  }
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/geoip-processor

3.8 user_agent Processor

User-Agent文字列を解析するProcessor

user_agent Processorは、HTTPヘッダーのUser-Agent文字列を解析し、ブラウザ名、OS名、デバイス情報などを抽出するProcessorです。Webアクセスログの分析や利用環境の可視化によく利用されます。

サンプルPipeline

PUT _ingest/pipeline/user-agent-pipeline
{
  "description": "User Agent Processor Sample",
  "processors": [
    {
      "user_agent": {
        "field": "agent"
      }
    }
  ]
}

実行例

POST sample-index/_doc?pipeline=user-agent-pipeline
{
  "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36"
}

実行結果 – Mapping

{
  "sample-index": {
    "mappings": {
      "properties": {
        "agent": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "user_agent": {
          "properties": {
            "device": {
              "properties": {
                "name": {
                  "type": "text",
                  "fields": {
                    "keyword": {
                      "type": "keyword",
                      "ignore_above": 256
                    }
                  }
                }
              }
            },
            "name": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "original": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            },
            "os": {
              "properties": {
                "full": {
                  "type": "text",
                  "fields": {
                    "keyword": {
                      "type": "keyword",
                      "ignore_above": 256
                    }
                  }
                },
                "name": {
                  "type": "text",
                  "fields": {
                    "keyword": {
                      "type": "keyword",
                      "ignore_above": 256
                    }
                  }
                },
                "version": {
                  "type": "text",
                  "fields": {
                    "keyword": {
                      "type": "keyword",
                      "ignore_above": 256
                    }
                  }
                }
              }
            },
            "version": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            }
          }
        }
      }
    }
  }
}

実行結果 – Data

{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "sample-index",
        "_id": "MwyKF58BsDOBCSPO-2lL",
        "_score": 1,
        "_source": {
          "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36",
          "user_agent": {
            "original": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36",
            "os": {
              "name": "Windows",
              "version": "10",
              "full": "Windows 10"
            },
            "name": "Chrome",
            "device": {
              "name": "Other"
            },
            "version": "137.0.0.0"
          }
        }
      }
    ]
  }
}

Reference

https://www.elastic.co/docs/reference/enrich-processor/user-agent-processor

enrich Processor

enrich processorについては別記事『Enrich』を参照ください。

Pipelineの実装・テスト・デバッグ

この章では、第3章で紹介した代表的なProcessorを組み合わせ、実際のログを加工するPipelineを作成します。

Pipelineの実装

実運用では1つのProcessorだけで要件を満たせることは少なく、

  • 複数のProcessorを組み合わせてログを構造化し、
  • 不要なFieldを削除し、
  • 分析しやすい形式へ変換すること

が一般的です。Elastic Certified Engineer Examでも、要件に応じて適切なProcessorを組み合わせたPipelineを構築できることが求められます。

以下は、アクセスログを解析するPipelineの例です。grokでログを解析し、dateで日時を変換し、geoipとuser_agentで情報を付加した後、set・rename・convert・removeを利用してDocumentを整理しています。

PUT _ingest/pipeline/access-log-pipeline
{
  "description": "Access Log Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{TIMESTAMP_ISO8601:log_time} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:response_time} \"%{GREEDYDATA:agent}\""
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": [
          "ISO8601"
        ]
      }
    },
    {
      "geoip": {
        "field": "client_ip"
      }
    },
    {
      "user_agent": {
        "field": "agent"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "message"
      }
    }
  ]
}

上記Pipelineでは、各Processorが次の役割を担っています。

Processor役割
grokアクセスログを各Fieldへ分割
date日時文字列をdate型へ変換
geoipIPアドレスから位置情報を付与
user_agentUser-Agentを解析
renamestatusをhttp_statusへ変更
convertresponse_timeをintegerへ変換
set共通Fieldを追加
remove不要になったmessageを削除

Processorは上から順番に実行されるため、実行順序は重要です。例えば、grokで生成したFieldをdateやgeoipが利用するため、grokは先頭で実行する必要があります。Processor間の依存関係を意識してPipelineを設計することが、実運用でもElastic Certified Engineer Examでも重要なポイントです。

Pipelineのテスト : _simulate API

Pipelineを作成したら、実際にIndexへ登録する前に期待どおりの変換結果になるか確認します。Ingest Pipelineでは、_simulate APIを利用することで、Documentを実際に登録することなくProcessorの実行結果を確認できます。Processorを追加・修正するたびに動作を確認できるため、実運用だけでなくElastic Certified Engineer Examでも積極的に活用したいAPIです。

作成済みのPipelineをテストする場合は、以下のように_simulate APIを実行します。

POST _ingest/pipeline/access-log-pipeline/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "202.232.2.100 2026-06-30T12:30:15 GET /products 200 125 \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""
      }
    }
  ]
}

実行すると、Pipelineを通過した後のDocumentが返されます。例えば、以下のような内容を確認できます。

  • grokによって各Fieldへ分割されていること
  • dateによって日時が変換されていること
  • geoipによって位置情報が付与されていること
  • user_agentによってブラウザ情報が解析されていること
  • rename・convert・set・removeが期待どおりに実行されていること

期待した結果にならない場合は、Processorの設定や実行順序を見直し、再度_simulate APIで確認します。DocumentをIndexへ登録せずに繰り返しテストできるため、安全かつ効率的にPipelineを開発できます。試験でも、複雑なPipelineを一度に作成するのではなく、Processorを段階的に追加・修正しながら_simulateで確認していく進め方がお勧めです。

Pipelineのデバッグ : on_failure

Elastic Certified Engineer Examでは、on_failureの詳細な挙動まで問われる問題が出題される可能性は高くありません。しかし、Ingest Pipelineを実運用するうえでは重要な機能であり、ProcessorレベルとPipelineレベルの違いや、エラー発生時の動作を理解しておくと役立ちます。本節では、実際にエラーを発生させながらon_failureの動作を確認していきます。

on_failureとは

実運用では、入力データの形式が想定と異なっていたり、Processorの設定に誤りがあったりすることで、Pipelineの実行中にエラーが発生することがあります。このような場合に利用できるのがon_failureです。on_failureは、Processorでエラーが発生した際に実行される例外処理であり、エラー内容をDocumentへ記録したり、代替処理を実行したりすることができます。

on_failureは、Pipeline全体と個々のProcessorの両方に定義できます。ただし、設定する場所によってエラー発生後の動作は異なります。

項目Pipelineレベルのon_failureProcessorレベルのon_failure
適用範囲Pipeline全体特定のProcessor
動作Pipeline全体の例外処理として動作指定したProcessorの例外処理として動作
エラー発生後の挙動通常のProcessorの実行は終了し、on_failureが実行されるon_failure実行後、後続のProcessorの実行を継続する

また、on_failureでは、エラー発生時の情報を取得するためのメタデータが提供されています。代表的なものとして、以下があります。

  • _ingest.on_failure_message : エラーメッセージを取得
  • _ingest.on_failure_processor_type : エラーが発生したProcessorの種類を取得

などがあります。これらを利用することで、エラー内容をDocumentへ保存し、後から原因を調査することができます。

次節では、PipelineレベルとProcessorレベルのon_failureについて、それぞれの動作の違いを確認していきます。

4.3.2 Pipelineレベルのon_failure

Pipelineレベルのon_failureは、Pipeline全体に対する例外処理です。いずれかのProcessorでエラーが発生すると、その時点で通常のProcessorの実行は終了し、on_failureで定義したProcessorが実行されます。そのため、エラーが発生した後続のProcessorは実行されません。一方で、on_failureが正常に完了すれば、加工済みのDocumentはIndexへ登録されます。

以下は、第4.1節で作成したPipelineへ、Pipelineレベルのon_failureを追加した例です。

PUT _ingest/pipeline/access-log-pipeline
{
  "description": "Access Log Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{TIMESTAMP_ISO8601:log_time} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:response_time} \"%{GREEDYDATA:agent}\""
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": [
          "ISO8601"
        ]
      }
    },
    {
      "geoip": {
        "field": "client_ip"
      }
    },
    {
      "user_agent": {
        "field": "agent"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "message"
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "error_information",
        "value": "Processor {{ _ingest.on_failure_processor_type }} failed with message: {{ _ingest.on_failure_message }}"
      }
    }
  ]
}

例えば、grok Processorでエラーが発生した場合、date、geoip、user_agentなど後続のProcessorは実行されません。その代わりにon_failureが実行され、error_informationへエラー内容が保存されます。

処理の流れは次のようになります。

grok
  ↓
エラー発生
  ↓
通常Processor終了
  ↓
Pipelineレベルのon_failure実行
  ↓
DocumentをIndexへ登録

一方、Pipelineレベルのon_failureを定義していない場合は、エラー発生時点でPipeline全体が失敗し、そのDocumentはIndexへ登録されません。

・pipelineにon_failureを含めた場合、errorが発生しても、データはindexに登録される
・pipelineにon_failureがない場合、errorが発生すると、そのデータはindexに登録されない

error発生時の挙動

次のコマンドで、Grokパターンと一致しないDocumentを登録し、実際にerrorを発生させてみます。

POST access-log/_doc?pipeline=access-log-pipeline
{
  "message": "invalid log format"
}

この例では、入力データはGrokパターンと一致していませんが、Pipelineレベルでon_failureを定義しているため、Documentの登録は正常に完了しています。そのため、レスポンスには"result": "created"、"_shards.successful": 1、"_shards.failed": 0が返されており、Indexへの登録が成功したことを確認できます。エラー内容はerror_information Fieldへ保存されているため、不正なデータを保持したまま後から原因を調査できます。

次に、登録されたデータを確認します。

GET access-log/_search

この例では、Grokパターンと一致しないログを登録したため、grok Processorでエラーが発生しています。しかし、Pipelineレベルでon_failureを定義しているため、Documentの登録自体は失敗せず、error_information Fieldへエラー内容が保存された状態でIndexへ登録されています。また、grokでエラーが発生した時点で通常のProcessorの実行は終了するため、dateやgeoip、user_agentなど後続のProcessorは実行されていません。

このように、on_failureを利用することで、不正なデータを破棄することなく、エラー内容を保持したまま後から原因を調査できます。

Processorレベルのon_failure

Processorレベルのon_failureは、特定のProcessorだけに対する例外処理です。Pipeline全体ではなく、指定したProcessorでのみエラーを補足します。そのため、on_failure実行後はPipelineの処理が継続され、後続のProcessorも実行されます。

以下は、convert Processorだけにon_failureを設定した例です。

PUT _ingest/pipeline/access-log-pipeline2
{
  "description": "Access Log Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{TIMESTAMP_ISO8601:log_time} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{GREEDYDATA:response_time} \"%{GREEDYDATA:agent}\""
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": [
          "ISO8601"
        ]
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer",
        "on_failure": [
          {
            "set": {
              "field": "response_time",
              "value": -1
            }
          }
        ]
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    }
  ]
}

この例では、response_timeが整数へ変換できない値だった場合でも、convert Processorのon_failureが実行され、response_timeへ-1が設定されます。その後は、後続のset Processorも通常どおり実行されます。

処理の流れは次のようになります。

grok
  ↓
date
  ↓
convertでエラー
  ↓
convertのon_failure実行
  ↓
set
  ↓
DocumentをIndexへ登録

このように、Processorレベルのon_failureは、エラーが発生したProcessorだけを補完し、Pipeline全体の処理を継続したい場合に利用します。一方、Pipelineレベルのon_failureは、Pipeline全体の例外処理として動作し、通常のProcessorの実行は終了します。

Elastic Certified Engineer Examでは、この2つのon_failureの違いが重要です。**Pipelineレベルは「Pipeline全体の例外処理」、Processorレベルは「Processor単位の例外処理」**であり、エラー発生後に後続Processorが実行されるかどうかが大きな違いとなります。

error発生時の挙動

次のコマンドで、convert processorでerrorを実行させてみます。

POST access-log/_doc?pipeline=access-log-pipeline2
{
  "message": "202.232.2.100 2026-06-30T12:30:15 GET /products 200 abc \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""
}

登録処理は、successful で完了します。

この例では、response_timeに数値へ変換できない値（abc）を指定したため、convert Processorでエラーが発生しています。しかし、convert Processorにon_failureを定義しているため、Pipeline全体は停止せず、response_timeへ-1が設定された後、後続のset Processorも実行されています。その結果、environment Fieldが追加された状態でDocumentが正常にIndexへ登録されていることが確認できます。

このように、Processorレベルのon_failureを利用すると、特定のProcessorで発生したエラーだけを補完し、Pipeline全体の処理を継続できます。

Bulk API実行時の挙動

ここまでは1件のDocumentを登録した場合のon_failureの動作を確認しました。実運用では、Bulk APIを利用して複数のDocumentをまとめて登録するケースが一般的です。この場合、あるDocumentでPipelineのエラーが発生した場合でも、Bulkリクエスト全体が中断されるわけではありません。各Documentは独立してPipelineが実行されるため、エラーが発生したDocumentだけが失敗し、正常なDocumentはそのままIndexへ登録されます。

以下は、正常なDocumentと異常なDocumentを同時に登録する例です。

POST _bulk
{"index":{"_index":"access-log","pipeline":"access-log-pipeline"}}
{"message":"202.232.2.100 2026-06-30T12:30:15 GET /products 200 125 \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""}
{"index":{"_index":"access-log","pipeline":"access-log-pipeline"}}
{"message":"invalid log format"}
{"index":{"_index":"access-log","pipeline":"access-log-pipeline"}}
{"message":"202.232.2.101 2026-06-30T12:31:20 POST /login 302 98 \"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""}

この例では、2件目のDocumentのみgrok Processorでエラーが発生します。

項目on_failureありon_failureなし
エラーが発生したDocumenton_failureが実行され、エラー情報を付与した状態で登録される登録されない
正常なDocumentすべて登録されるすべて登録される
登録件数（3件投入した場合）3件2件
Bulk処理最後まで継続される最後まで継続される

つまり、Bulk APIでは1件のDocumentの失敗によって後続のDocumentの登録が中断されることはありません。PipelineのエラーはDocument単位で判定されるため、失敗したDocumentのみが影響を受け、その他のDocumentは通常どおり処理されます。

on_failureなしのpipelineを介してBulk登録

次に、on_failureを定義していないPipelineを利用し、エラーとなるDocumentを含むBulk APIを実行してみます。この場合、エラーが発生したDocumentは登録されませんが、他の正常なDocumentは影響を受けず、そのままIndexへ登録されます。Bulk API実行時の挙動を確認してみましょう。

on_failureなしpipeline生成コマンド

PUT _ingest/pipeline/access-log-pipeline-no-on_failure
{
  "description": "Access Log Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{TIMESTAMP_ISO8601:log_time} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:response_time} \"%{GREEDYDATA:agent}\""
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": [
          "ISO8601"
        ]
      }
    },
    {
      "geoip": {
        "field": "client_ip"
      }
    },
    {
      "user_agent": {
        "field": "agent"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "message"
      }
    }
  ]
}

bulkコマンド

POST _bulk
{"index":{"_index":"access-log","pipeline":"access-log-pipeline-no-on_failure"}}
{"message":"202.232.2.100 2026-06-30T12:30:15 GET /products 200 125 \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""}
{"index":{"_index":"access-log","pipeline":"access-log-pipeline-no-on_failure"}}
{"message":"invalid log format"}
{"index":{"_index":"access-log","pipeline":"access-log-pipeline-no-on_failure"}}
{"message":"202.232.2.101 2026-06-30T12:31:20 POST /login 302 98 \"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""}

結果

2件目のデータが”status”: 400でerrorとなっています。

indexに登録されたデータも、3件中2件となります。

実践問題

Practice Problem 1

You need to create an ingest pipeline named basic-pipeline.

The pipeline must satisfy the following requirements:

  • Rename the field status to http_status.
  • Convert the field response_time to an integer.
  • Add a field named environment with the value production.
  • Remove the field message.

After creating the pipeline, ingest the following document using the pipeline.

{
  "status": "200",
  "response_time": "125",
  "message": "GET /products"
}

回答

Pipelineの作成

PUT _ingest/pipeline/basic-pipeline
{
  "description": "Basic Pipeline",
  "processors": [
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "message"
      }
    }
  ]
}

Documentの登録

POST sample-index/_doc?pipeline=basic-pipeline
{
  "status": "200",
  "response_time": "125",
  "message": "GET /products"
}

解説

  • “rename”で”status”を”http_status”へ変更します。
  • “convert”で”response_time”を文字列から”integer”型へ変換します。
  • “set”で”environment=production”を追加します。
  • “remove”で不要な”message”を削除します。
  • 作成したPipelineは、”pipeline”パラメータを指定してDocumentを登録することで適用されます。

Practice Problem 2

You need to create an ingest pipeline named access-log-pipeline.

The pipeline must satisfy the following requirements:

  • Parse the message field using the grok processor.
  • Extract the following fields:
    • client_ip
    • log_time
    • method
    • request
    • status
    • response_time
    • agent
  • Convert log_time to a date field.
  • Add GeoIP information based on client_ip.
  • Parse the agent field using the user_agent processor.
  • Rename status to http_status.
  • Convert response_time to an integer.
  • Add a field named environment with the value production.
  • Remove the original message field.

After creating the pipeline, ingest the following document using the pipeline.

{
  "message": "202.232.2.100 2026-06-30T12:30:15 GET /products 200 125 \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""
}

回答

Pipelineの作成

PUT _ingest/pipeline/access-log-pipeline
{
  "description": "Access Log Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} %{TIMESTAMP_ISO8601:log_time} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:response_time} \"%{GREEDYDATA:agent}\""
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": [
          "ISO8601"
        ]
      }
    },
    {
      "geoip": {
        "field": "client_ip"
      }
    },
    {
      "user_agent": {
        "field": "agent"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "message"
      }
    }
  ]
}

Documentの登録

POST access-log/_doc?pipeline=access-log-pipeline
{
  "message": "202.232.2.100 2026-06-30T12:30:15 GET /products 200 125 \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36\""
}

解説

  • “grok”でアクセスログを解析し、IPアドレス、日時、HTTPメソッド、リクエストパス、ステータスコード、レスポンス時間、User-Agentを抽出します。
  • “date”で”log_time”をdate型へ変換します。
  • “geoip”で”client_ip”から位置情報を付与します。
  • “user_agent”でブラウザやOSなどの情報を解析します。
  • “rename”で”status”を”http_status”へ変更します。
  • “convert”で”response_time”を”integer”型へ変換します。
  • “set”で”environment”に”production”を設定します。
  • “remove”で不要になった”message”を削除します。

Practice Problem 3

You need to create an ingest pipeline named client-info-pipeline.

The pipeline must satisfy the following requirements:

  • Add GeoIP information based on the client_ip field.
  • Parse the agent field using the user_agent processor.
  • Rename the field status to http_status.
  • Convert the field response_time to an integer.
  • Add a field named environment with the value production.
  • Remove the field temp_field.

After creating the pipeline, ingest the following document using the pipeline.

{
  "client_ip": "202.232.2.100",
  "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36",
  "status": "200",
  "response_time": "125",
  "temp_field": "temporary data"
}

回答

PUT _ingest/pipeline/client-info-pipeline
{
  "description": "Client Info Pipeline",
  "processors": [
    {
      "geoip": {
        "field": "client_ip"
      }
    },
    {
      "user_agent": {
        "field": "agent"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "temp_field"
      }
    }
  ]
}
POST client-info/_doc?pipeline=client-info-pipeline
{
  "client_ip": "202.232.2.100",
  "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/137.0.0.0 Safari/537.36",
  "status": "200",
  "response_time": "125",
  "temp_field": "temporary data"
}

解説

  • geoip client_ipから国・都市・緯度経度などの位置情報を付与します。
  • user_agent agentを解析し、ブラウザ名、OS名、デバイス情報などを抽出します。
  • rename statusをhttp_statusへ変更し、Field名を分かりやすく統一します。
  • convert response_timeを文字列からinteger型へ変換します。
  • set environmentへproductionを設定し、共通情報をDocumentへ追加します。
  • remove 不要なtemp_fieldを削除し、Documentを整理します。

Practice Problem 4

You need to create an ingest pipeline named date-pipeline.

The pipeline must satisfy the following requirements:

  • Convert the field log_time to the @timestamp field.
  • The input date format is yyyy/MM/dd HH:mm:ss.
  • Keep the original log_time field.
  • Add a field named environment with the value production.

After creating the pipeline, ingest the following document using the pipeline.

{
  "log_time": "2026/06/30 12:30:15",
  "status": "200",
  "response_time": 125
}

回答

Pipelineの作成

PUT _ingest/pipeline/date-pipeline
{
  "description": "Date Pipeline",
  "processors": [
    {
      "date": {
        "field": "log_time",
        "target_field": "@timestamp",
        "formats": [
          "yyyy/MM/dd HH:mm:ss"
        ]
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    }
  ]
}

Documentの登録

POST access-log/_doc?pipeline=date-pipeline
{
  "log_time": "2026/06/30 12:30:15",
  "status": "200",
  "response_time": 125
}

解説

  • “date” “log_time”を指定したフォーマット（”yyyy/MM/dd HH:mm:ss”）で解析し、”@timestamp”へ変換します。
  • “target_field” 変換後の日時を格納するFieldを指定します。省略した場合は”@timestamp”が使用されますが、明示的に指定することで意図が分かりやすくなります。
  • “formats” 入力される日時文字列の形式を指定します。複数のフォーマットを指定することも可能です。
  • “set” “environment”に”production”を設定します。

Practice Problem 5

You need to create an ingest pipeline named url-access-pipeline.

The pipeline must satisfy the following requirements:

  • Parse the url field using a processor that extracts URL components.
  • Store the parsed URL components under the field url_parts.
  • Rename the field status to http_status.
  • Convert the field response_time to an integer.
  • Add a field named environment with the value production.
  • Remove the field temp_field.

After creating the pipeline, ingest the following document using the pipeline.

{
  "url": "<https://www.example.com/products?id=100>",
  "status": "200",
  "response_time": "125",
  "temp_field": "temporary data"
}

回答

Pipelineの作成

PUT _ingest/pipeline/url-access-pipeline
{
  "description": "URL Access Pipeline",
  "processors": [
    {
      "uri_parts": {
        "field": "url",
        "target_field": "url_parts"
      }
    },
    {
      "rename": {
        "field": "status",
        "target_field": "http_status"
      }
    },
    {
      "convert": {
        "field": "response_time",
        "type": "integer"
      }
    },
    {
      "set": {
        "field": "environment",
        "value": "production"
      }
    },
    {
      "remove": {
        "field": "temp_field"
      }
    }
  ]
}

Documentの登録

POST access-log/_doc?pipeline=url-access-pipeline
{
  "url": "<https://www.example.com/products?id=100>",
  "status": "200",
  "response_time": "125",
  "temp_field": "temporary data"
}

解説

  • “uri_parts”
    • 第3章では扱っていないProcessorです。
    • URL文字列を解析し、scheme、domain、path、queryなどの構成要素へ分割します。
    • 試験で見慣れないProcessorが出題された場合でも、公式ドキュメントを参照し、Processor名、必須Field、主要Optionを確認して実装することが重要です。
  • “rename” “status”を”http_status”へ変更します。
  • “convert” “response_time”を文字列から”integer”型へ変換します。
  • “set” “environment”に”production”を設定します。
  • “remove” 不要な”temp_field”を削除します。

Elastic社が提供する無料Hands-on Trainingで理解を深めよう

ここまで、Ingest Pipelineの基本構成から代表的なProcessor、Pipelineの実装・テスト・デバッグ方法について解説しました。知識を定着させるためには、実際にPipelineを作成し、Documentを登録・加工しながら動作を確認することが重要です。

Elastic社では、無料で利用できるHands-on TrainingやOn-Demand Trainingを提供しています。これらの教材では、Ingest Pipelineをはじめ、Index TemplateやData Stream、Elastic Agentなどと組み合わせた実践的なシナリオを体験できるため、試験対策だけでなく実運用を見据えた理解を深めることができます。

特にIngest Pipelineは、Processor単体の使い方だけでなく、複数のProcessorを組み合わせたPipelineの設計や、_simulate APIを利用したテスト、エラー発生時のデバッグ方法まで実際に手を動かしながら学習することをお勧めします。実践を通じて理解を深めることで、Elastic Certified Engineer Examで求められる「要件に応じたPipelineの実装力」を身につけられるでしょう。

Elastic Cloud アカウント作成

https://cloud.elastic.co/registration

Elastic Training

https://www.elastic.co/training

Elastic Learning Portal

https://learn.elastic.co/pages/58/home-page

試験で問われるポイント

  • 要件に応じて適切なProcessorを選択し、Ingest Pipelineを実装できること
  • set、rename、remove、convert、date、grok、geoip、user_agentなど、代表的なProcessorの役割と使い分けを理解していること
  • Pipeline内のProcessorは定義した順番に実行されることを理解し、Processor間の依存関係を考慮したPipelineを設計できること
  • _simulate APIを利用してPipelineをテストし、期待どおりの変換結果になっているか確認・デバッグできること
  • 作成したPipelineを、Index Templateのdefault_pipelineとして設定し、自動適用できること
  • on_failureの詳細な挙動まで問われる可能性は高くありませんが、基本的な動作は理解しておくこと
  • grokパターンを暗記する必要はありませんが、利用可能なパターンを一覧表示するコマンド(下部に記載)を覚えておくこと
  • 主要なProcessor以外についても、どのようなProcessorが用意されているかを把握しておき、見慣れないProcessorが出題された場合でも、公式マニュアルを参照しながら実装できるようにしておくこと
GET _ingest/processor/grok?s