Elastic Certified Engineer Exam対策 – Reindex & Update By Query

Banner for a training course: 'Elastic Certified Engineer Exam 対策' with 'TRAINING' vertical label on the right. トレーニング

このブログの目的

本記事では、Elastic Certified Engineer試験の試験範囲である以下の項目について解説します。

Data Processing – Use the Reindex API and Update By Query API to reindex and/or update documents

Reindex APIは、新しいIndexへのデータ移行だけでなく、Queryによるコピー対象の絞り込みや、Script・Ingest Pipelineを利用したデータ加工を行いながらデータを再構築するためのAPIです。一方、Update By Query APIは、Queryに一致する複数のドキュメントを一括更新し、Painless Scriptを利用してFieldの追加・更新・削除や条件分岐を実装するためのAPIです。どちらも実運用で利用頻度が高く、Elastic Certified Engineer試験でも重要な出題範囲となっています。

本記事では、Reindex APIとUpdate By Query APIの基本的な使い方に加え、Query・Script・Ingest Pipelineを組み合わせた実践的な利用方法や、Update APIとの違い、非同期実行とTask APIの活用方法について解説します。また、章末には試験を想定した実践問題も掲載していますので、知識の定着や試験対策に役立てていただければ幸いです。

試験情報は以下サイトで確認可能です。

https://www.elastic.co/training/elastic-certified-engineer-exam

Reindex API

Reindex APIの基本

Reindex APIは、既存のIndexに格納されているドキュメントを、新しいIndexへコピーするためのAPIです。単にデータを複製するだけでなく、Queryによるコピー対象の絞り込みや、Script・Ingest Pipelineを利用したデータ加工を行いながら、新しいIndexを作成できることが大きな特徴です。

実運用では、Mappingの変更やIndex設計の見直し、新しいIndexへのデータ移行などで広く利用されており、Elastic Certified Engineer試験でも頻出のAPIです。

今回は、サンプルデータとして「kibana_sample_data_ecommerce」Indexを利用し、「kibana_sample_data_ecommerce_reindexed」という新しいIndexへデータをコピーしてみます。

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_reindexed"
  }
}

実行すると、コピー元Indexのドキュメントがコピー先Indexへ移行されます。なお、Reindex APIはドキュメントのみをコピーするAPIであり、MappingやSettingsはコピーされません。

コピー先Indexが存在しない場合は自動的に作成されますが、そのMappingはDynamic Mappingや適用されるIndex Templateによって決定されます。そのため、コピー元と同じMappingを利用したい場合は、事前にMappingを取得し、コピー先Indexを作成しておくことを推奨します。

Queryを利用したReindex

前節では、コピー元Indexに格納されているすべてのドキュメントを新しいIndexへコピーしました。しかし、実際の運用では、特定の条件に一致するドキュメントだけをコピーしたいケースも少なくありません。

Reindex APIでは、”source” に “query” を指定することで、コピー対象のドキュメントを絞り込むことができます。利用できるQueryはSearch APIと同じであり、Term QueryやRange Query、Bool Queryなど、Query DSLで利用できるさまざまなQueryを指定できます。

例えば、”customer_gender” が “”MALE”” のドキュメントのみを新しいIndexへコピーする場合は、以下のように実装します。

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce",
    "query": {
      "term": {
        "customer_gender": "MALE"
      }
    }
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_reindexed"
  }
}

この例では、”customer_gender” が “”MALE”” のドキュメントのみが “kibana_sample_data_ecommerce_reindexed” へコピーされます。””FEMALE”” のドキュメントはコピーされません。

事前に以下のクエリを実行すると、”customer_gender” ごとのドキュメント数を確認できます。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "aggs": {
    "customer_gender": {
      "terms": {
        "field": "customer_gender"
      }
    }
  }
}

“query” には複雑な条件を指定することもできます。例えば、複数条件を組み合わせたBool Queryや、期間を指定するRange Queryなども利用できます。そのため、必要なデータだけを抽出しながら、新しいIndexを作成することが可能です。

Queryを利用したReindexは、実運用でも非常によく利用される機能です。例えば、特定期間のデータだけを別Indexへ移行したり、特定の条件に一致するデータのみをアーカイブしたりする際に活用されます。また、Elastic Certified Engineer試験でも、Queryを利用して要件を満たすドキュメントだけをReindexする問題が出題される可能性が高いため、基本的な実装方法は確実に習得しておきましょう。

筆者が受験した際には、まずQueryで条件に一致するドキュメントを検索し、その後、Reindexで新しいIndexを作成して、両者の件数が一致することを確認する問題が出題されました。

設問には「Reindexでも同じQueryを利用する」とは明記されていなかったため、一見すると件数が一致しないように思えます。しかし、要件を満たすには、Reindexのsource.queryにも同じQueryを指定し、コピー対象を絞り込む必要があります。

試験では、このように設問に明記されていない処理を、要件から読み取って実装する力が求められる場合があります。

筆者が受験した際には、まずQueryで条件に一致するドキュメントを検索し、その後、Reindexで新しいIndexを作成して、両者の件数が一致することを確認する問題が出題されました。設問には「Reindexでも同じQueryを利用する」とは明記されていなかったため、一見すると件数が一致しないように思えます。しかし、要件を満たすには、Reindexのsource.queryにも同じQueryを指定し、コピー対象を絞り込む必要があります。試験では、このように設問に明記されていない処理を、要件から読み取って実装する力が求められる場合があります。

Scriptを利用したReindex

前節では、Queryを利用してコピー対象のドキュメントを絞り込みました。Reindex APIではさらに、Scriptを利用することで、ドキュメントを加工しながら新しいIndexへコピーできます。

例えば、不要になった “customer_phone” Fieldを削除しながらReindexする場合は、以下のように実装します。

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_reindexed"
  },
  "script": {
    "source": "ctx._source.remove('customer_phone')"
  }
}

実行すると、”customer_phone” Fieldを削除した状態でドキュメントが “kibana_sample_data_ecommerce_reindexed” へコピーされます。一方、コピー元の “kibana_sample_data_ecommerce” Indexには影響はなく、Fieldもそのまま保持されます。

Scriptでは、Fieldの削除だけでなく、Fieldの追加や更新、Field名の変更なども実現できます。そのため、Mappingの変更に伴うデータ加工や、不要なFieldの整理など、実運用でも幅広く利用されています。

Ingest Pipelineを利用したReindex

前節では、Scriptを利用してドキュメントを加工しながらReindexする方法を紹介しました。Reindex APIでは、Scriptだけでなく、Ingest Pipelineを利用してデータを加工しながら新しいIndexへコピーすることもできます。

Ingest Pipelineを利用する場合は、”dest.pipeline” にPipeline名を指定します。Reindex実行時に、コピー先Indexへドキュメントを書き込む前にPipelineが実行されるため、Processorによるデータ加工を行うことができます。

まずは、Reindex時に利用するPipelineを作成します。

PUT _ingest/pipeline/ecommerce-reindex-pipeline
{
  "description": "Pipeline for reindexing ecommerce sample data",
  "processors": [
    {
      "set": {
        "field": "reindexed",
        "value": true
      }
    },
    {
      "rename": {
        "field": "customer_full_name",
        "target_field": "customer_name",
        "ignore_missing": true
      }
    },
    {
      "remove": {
        "field": "customer_phone",
        "ignore_missing": true
      }
    }
  ]
}

このPipelineでは、以下の処理を行っています。

Processor処理内容
set“reindexed” Fieldを追加し、値に “true” を設定
rename“customer_full_name” Fieldを “customer_name” に変更
remove“customer_phone” Fieldを削除

次に、作成したPipelineを指定してReindexを実行します。

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_reindexed",
    "pipeline": "ecommerce-reindex-pipeline"
  }
}

この例では、”kibana_sample_data_ecommerce” のドキュメントが、”ecommerce-reindex-pipeline” を経由してから “kibana_sample_data_ecommerce_reindexed” に書き込まれます。その結果、コピー先Indexでは、”reindexed” Fieldが追加され、”customer_full_name” Fieldは “customer_name” に変更され、”customer_phone” Fieldは削除されます。

Update By Query API

Update By Query APIの基本

Update By Query APIは、既存のIndexに格納されているドキュメントのうち、指定した条件に一致するドキュメントを一括更新するためのAPIです。単一のドキュメントを更新するUpdate APIとは異なり、Queryによって更新対象を絞り込み、複数のドキュメントをまとめて更新できることが大きな特徴です。

また、Painless Scriptを利用することで、Fieldの追加・更新・削除などのデータ加工を行いながら更新することもできます。実運用では、既存データの一括補正やField値の変換などで利用されることが多く、Elastic Certified Engineer試験でも理解しておくべきAPIの一つです。

今回は、サンプルデータとして「kibana_sample_data_ecommerce」Indexを利用し、すべてのドキュメントに「updated」というFieldを追加してみます。

POST kibana_sample_data_ecommerce/_update_by_query
{
  "script": {
    "source": "ctx._source.updated = true"
  }
}

実行すると、「kibana_sample_data_ecommerce」に格納されているすべてのドキュメントに「updated」Fieldが追加され、値としてtrueが設定されます。以降の節では、Queryによる更新対象の絞り込みや、Scriptを利用したより実践的な更新方法について解説します。

Queryを利用したUpdate By Query

前節では、Index内のすべてのドキュメントを更新しました。しかし、実際の運用では、特定の条件に一致するドキュメントだけを更新したいケースも少なくありません。

Update By Query APIでは、”query”を指定することで、更新対象のドキュメントを絞り込むことができます。利用できるQueryはSearch APIと同じであり、Term QueryやRange Query、Bool Queryなど、Query DSLで利用できるさまざまなQueryを指定できます。

例えば、”customer_gender”が”MALE”のドキュメントだけを更新し、「updated」Fieldの値をfalseに変更する場合は、以下のように実装します。

POST kibana_sample_data_ecommerce/_update_by_query
{
  "query": {
    "term": {
      "customer_gender": "MALE"
    }
  },
  "script": {
    "source": "ctx._source.updated = false"
  }
}

この例では、”customer_gender”が”MALE”のドキュメントのみが更新され、「updated」Fieldの値がfalseになります。”FEMALE”のドキュメントは更新されません。

事前に以下のクエリを実行すると、”customer_gender”ごとのドキュメント数を確認できます。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "aggs": {
    "customer_gender": {
      "terms": {
        "field": "customer_gender"
      }
    }
  }
}

“query”には複雑な条件を指定することもできます。例えば、Bool QueryやRange Queryなどを利用することで、必要なドキュメントだけを対象に更新できます。実運用では、特定期間のデータだけを補正したり、条件に一致するドキュメントのみを一括更新したりする場面でよく利用されます。また、Elastic Certified Engineer試験でも、Queryを利用して要件を満たすドキュメントだけを更新する問題が出題される可能性があるため、基本的な実装方法は確実に習得しておきましょう。

Scriptを利用したField更新

前節では、Queryを利用して更新対象のドキュメントを絞り込みました。Update By Query APIではさらに、Painless Scriptを利用することで、Fieldの追加・更新・削除だけでなく、条件分岐や計算処理など、より柔軟なデータ加工を行いながらドキュメントを更新できます。

例えば、”taxful_total_price”の値に応じて、価格帯を表す「price_category」Fieldを追加する場合は、以下のように実装します。

POST kibana_sample_data_ecommerce/_update_by_query
{
  "script": {
    "source": """
      if (ctx._source.taxful_total_price >= 200) {
        ctx._source.price_category = "HIGH";
      } else {
        ctx._source.price_category = "NORMAL";
      }
    """
  }
}

この例では、”taxful_total_price”が200以上のドキュメントには”price_category” Fieldに”HIGH”が、それ以外のドキュメントには”NORMAL”が設定されます。

aggsでtext typeを指定するとerrorとなる

この例では、新しく追加される「price_category」Fieldは、Dynamic Mappingによって自動的にMappingが作成されます。文字列Fieldとして作成された場合、Terms Aggregationなどで「price_category」を直接指定すると、Fielddata is disabledというエラーが発生することがあります。

その場合は、集計用のサブフィールドである「price_category.keyword」を指定してください。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "aggs": {
    "price_category": {
      "terms": {
        "field": "price_category.keyword"
      }
    }
  }
}

なお、これはUpdate By Query API固有の仕様ではなく、text型FieldをAggregationやSortに利用する際のElasticsearch共通の仕様です。

Update APIとの違い

Update APIとUpdate By Query APIは、どちらも既存のドキュメントを更新するAPIですが、更新対象が大きく異なります。

Update APIは、Document IDを指定して1件のドキュメントを更新するAPIです。一方、Update By Query APIは、Queryに一致した複数のドキュメントを一括更新するAPIです。そのため、更新したいドキュメントが1件であることが分かっている場合はUpdate APIを、条件に一致する複数のドキュメントをまとめて更新したい場合はUpdate By Query APIを利用します。

例えば、Document IDがVvJsmJcB5H4mJ7mH9xKfのドキュメントに「updated」Fieldを追加する場合は、以下のようにUpdate APIを実行します。

POST kibana_sample_data_ecommerce/_update/VvJsmJcB5H4mJ7mH9xKf
{
  "script": {
    "source": "ctx._source.updated = true"
  }
}

このように、Update APIは更新対象のDocument IDが分かっている場合に利用します。一方、Document IDが分からず、「customer_genderが”MALE”のドキュメント」など、条件に一致する複数のドキュメントを更新したい場合は、Update By Query APIを利用します。

両者の主な違いを以下に示します。

Update APIUpdate By Query API
更新対象は1件のドキュメント更新対象はQueryに一致する複数のドキュメント
Document IDを指定して更新Queryで更新対象を指定
エンドポイント：POST index/_update/{id}エンドポイント：POST index/_update_by_query
単一ドキュメントの更新に適している一括更新に適している

wait_for_completion=false と Task API

Update By Query APIは、大量のドキュメントを更新する場合、処理に時間がかかることがあります。そのような場合は、wait_for_completion=falseを指定することで、更新処理を非同期で実行できます。

POST kibana_sample_data_ecommerce/_update_by_query?wait_for_completion=false
{
  "script": {
    "source": "ctx._source.updated = true"
  }
}

実行すると、更新処理が完了するのを待たずにレスポンスが返され、Task IDが返却されます。

{
  "task": "Zt5YAm63QQqDupFLLjpnZg:4881547"
}

返却されたTask IDは、Task APIを利用して進捗状況を確認できます。

GET _tasks/Zt5YAm63QQqDupFLLjpnZg:4881547

処理が完了すると、レスポンスのcompletedがtrueとなり、更新件数などの実行結果を確認できます。

なお、Task IDは実行環境によって異なります。実行する際は、ご自身の環境で返却されたTask IDを指定してください。

大量データを対象としたUpdate By Queryでは、処理時間が長くなることがあります。
wait_for_completion=falseとTask APIを利用することで、処理をバックグラウンドで実行しながら進捗を確認できるため、実運用でも広く利用される機能です。

実践問題

Question 1

You need to create a new index that contains only the documents whose customer_gender is MALE.

Create a new index named kibana_sample_data_ecommerce_male by using the Reindex API.

回答

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce",
    "query": {
      "term": {
        "customer_gender": "MALE"
      }
    }
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_male"
  }
}

解説

この問題では、Reindex APIでsource.queryを指定し、コピー対象のドキュメントを絞り込めるかが問われています。

  • source.indexにはコピー元のIndexを指定します。
  • dest.indexにはコピー先のIndexを指定します。
  • source.queryにTerm Queryを指定することで、customer_genderがMALEのドキュメントのみを新しいIndexへコピーできます。

Reindex実行後は、以下のAggregationを実行することで、コピー先Indexに”MALE”のドキュメントのみが格納されていることを確認できます。

GET kibana_sample_data_ecommerce_male/_search
{
  "size": 0,
  "aggs": {
    "customer_gender": {
      "terms": {
        "field": "customer_gender"
      }
    }
  }
}

レスポンスに”MALE”のみが表示されれば、期待どおりにReindexできています。

Question 2

You need to create a new index for analysis, but the original documents contain a field that should not be included in the new index.

Create a new index named kibana_sample_data_ecommerce_without_currency by using the Reindex API. During reindexing, remove the currency field from each copied document.

回答

POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_without_currency"
  },
  "script": {
    "source": "ctx._source.remove('currency')"
  }
}

解説

この問題では、Reindex APIでScriptを利用し、ドキュメントを加工しながら新しいIndexへコピーできるかが問われています。

  • source.indexにはコピー元のIndexを指定します。
  • dest.indexにはコピー先のIndexを指定します。
  • script.sourceにctx._source.remove('currency')を指定することで、Reindex中にコピー対象のDocumentからcurrency Fieldを取り除いたうえで、コピー先Indexへ書き込むことができます(コピー元に影響はありません)。

実行後は、以下のクエリでコピー先Indexにcurrency Fieldが存在しないことを確認できます。

GET kibana_sample_data_ecommerce_without_currency/_search
{
  "query": {
    "exists": {
      "field": "currency"
    }
  }
}

hits.total.valueが0であれば、currency Fieldが削除された状態でReindexできています。

Question 3

You need to create a new index for reporting purposes.

Create an ingest pipeline named ecommerce-reporting-pipeline that performs the following transformations:

  • Add a field named reporting_index with the value true
  • Rename customer_full_name to customer_name
  • Remove the customer_phone field

Then, create a new index named kibana_sample_data_ecommerce_reporting by using the Reindex API with this pipeline.

回答

PUT _ingest/pipeline/ecommerce-reporting-pipeline
{
  "description": "Pipeline for ecommerce reporting index",
  "processors": [
    {
      "set": {
        "field": "reporting_index",
        "value": true
      }
    },
    {
      "rename": {
        "field": "customer_full_name",
        "target_field": "customer_name",
        "ignore_missing": true
      }
    },
    {
      "remove": {
        "field": "customer_phone",
        "ignore_missing": true
      }
    }
  ]
}
POST _reindex
{
  "source": {
    "index": "kibana_sample_data_ecommerce"
  },
  "dest": {
    "index": "kibana_sample_data_ecommerce_reporting",
    "pipeline": "ecommerce-reporting-pipeline"
  }
}

解説

この問題では、Ingest Pipelineを作成し、そのPipelineをReindex時に適用できるかが問われています。

  • dest.pipelineには、Reindex時に利用するPipeline名を指定します。
  • 指定したPipelineは、コピー元Indexから取得したDocumentがコピー先Indexへ書き込まれる前に実行されます。
  • set Processorでは、reporting_index Fieldを追加し、値にtrueを設定しています。
  • rename Processorでは、customer_full_name Fieldをcustomer_nameへ変更しています。
  • remove Processorでは、customer_phone Fieldを削除した状態でコピー先Indexへ書き込みます。

実行後は、以下のようにコピー先Indexのドキュメントを確認すると、Pipelineによる変換結果を確認できます。

GET kibana_sample_data_ecommerce_reporting/_search
{
  "size": 1,
  "_source": [
    "reporting_index",
    "customer_name",
    "customer_full_name",
    "customer_phone"
  ]
}

レスポンスでreporting_indexとcustomer_nameが存在し、customer_full_nameとcustomer_phoneが存在しなければ、期待どおりにPipelineを利用したReindexができています。

Question 4

The business has introduced a new order classification.

Add a new field named high_value_order to all documents in the kibana_sample_data_ecommerce index.

Set high_value_order to true when taxful_total_price is greater than or equal to 200.

Set high_value_order to false when taxful_total_price is less than 200.

Write the Elasticsearch API request to perform this update.

回答

POST kibana_sample_data_ecommerce/_update_by_query
{
  "script": {
    "source": """
      if (ctx._source.taxful_total_price >= 200) {
        ctx._source.high_value_order = true;
      } else {
        ctx._source.high_value_order = false;
      }
    """,
    "lang": "painless"
  }
}

解説

この問題では、すべてのDocumentを対象にして、taxful_total_priceの値に応じてhigh_value_orderを設定します。

  • ctx._sourceは、更新対象となるDocumentの_sourceを表します。
  • if文を使用し、taxful_total_priceが200以上の場合はhigh_value_orderにtrueを設定し、それ以外の場合はfalseを設定しています。
  • queryを指定していないため、Index内のすべてのDocumentが更新対象となります。

更新後は、以下のクエリで確認できます。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "aggs": {
    "high_value_order_count": {
      "terms": {
        "field": "high_value_order"
      }
    }
  }
}

Question 5

Write an Update By Query API request that updates all documents in the kibana_sample_data_ecommerce index asynchronously.

Add a new field named campaign_target.

Set campaign_target to true when all of the following conditions are met.

  • customer_gender is FEMALE
  • taxful_total_price is greater than or equal to 100
  • day_of_week is either Saturday or Sunday

Otherwise, set campaign_target to false.

回答

POST kibana_sample_data_ecommerce/_update_by_query?wait_for_completion=false
{
  "script": {
    "source": """
      if (
        ctx._source.customer_gender == 'FEMALE' &&
        ctx._source.taxful_total_price >= 100 &&
        (ctx._source.day_of_week == 'Saturday' || ctx._source.day_of_week == 'Sunday')
      ) {
        ctx._source.campaign_target = true;
      } else {
        ctx._source.campaign_target = false;
      }
    """,
    "lang": "painless"
  }
}

解説

この問題では、Update By Query APIを非同期で実行し、Painless Scriptで条件分岐を実装できるかが問われています。

  • wait_for_completion=falseを指定することで、更新処理の完了を待たずにTask IDを返却させています。
  • queryを指定していないため、Index内のすべてのDocumentが更新対象となります。
  • if文を使用し、customer_gender、taxful_total_price、day_of_weekの3つの条件をすべて満たす場合はcampaign_targetにtrueを、それ以外の場合はfalseを設定しています。
  • 複数の条件は、論理積演算子&&と論理和演算子||を組み合わせて実装します。

非同期で実行した場合、レスポンスにはTask IDが返却されます。このTask IDを利用することで、Task APIから更新処理の進捗や完了状態を確認できます。

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

本記事では、Reindex APIおよびUpdate By Query APIの基本的な使い方や、Query・Script・Ingest Pipelineを利用した実践的な活用方法について解説しました。これらの機能は、実際に手を動かしながら試すことで理解が深まります。

Elastic社では、無料で受講できるHands-on Trainingを提供しています。演習環境があらかじめ用意されているため、Elasticsearchをインストールすることなく、ブラウザから実際にAPIを実行しながら学習できます。

特にElastic Certified Engineer試験を受験予定の方は、Dev Toolsを使ってReindex APIやUpdate By Query APIを何度も実装し、各APIの使い分けやパラメータを自然に使えるようになるまで繰り返し演習することをおすすめします。

なお、無料のHands-on Trainingを利用するには、Elastic Cloudのアカウントが必要です。アカウントは無料で作成でき、Hands-onの受講や演習環境へのアクセスに利用します。事前にアカウントを作成しておくことをおすすめします。

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

試験で問われるポイント

  • Reindex APIは、Query・Script・Ingest Pipelineを組み合わせて、新しいIndexへデータをコピーできるようにしておく。特に、source.queryやdest.pipelineの指定方法は頻出である。
  • Update By Query APIは、Queryで更新対象を絞り込み、Painless Scriptを利用してFieldの追加・更新・削除や条件分岐を実装できるようにしておく。ctx._sourceやif文の基本的な記述方法は習得しておきたい。

Reindex APIとUpdate By Query APIは、出題される可能性が高い重要なトピックです。基本的な実装方法や使い分けを理解し、確実に解答できるようにしておきましょう。