チュートリアル: eCommerce サンプルデータの変換

Transforms を使用すると、Elasticsearch インデックスから情報を取得し、それを変換して別のインデックスに保存できます。ここでは、Kibana サンプルデータ を使用して、変換を使ってデータをピボットし、要約する方法を示します。

  • 1. 環境が変換を使用するために正しく設定されていることを確認します。Elasticsearch のセキュリティ機能が有効になっている場合、このチュートリアルを完了するには、変換をプレビューおよび作成する権限を持つユーザーが必要です。また、ソースおよび宛先インデックスに対して特定のインデックス権限を持っている必要があります。詳細は Setup を参照してください。
  • 2. ソースインデックスを選択します。
    この例では、eCommerce 注文のサンプルデータを使用します。kibana_sample_data_ecommerce インデックスにまだ慣れていない場合は、Kibana の Revenue ダッシュボードを使用してデータを探索してください。この eCommerce データからどのような洞察を得たいかを考えてみてください。
  • 3. 変換のピボットタイプを選択し、データのグループ化や集約のさまざまなオプションを試してみてください。
    変換には 2 種類のタイプがありますが、まずはデータをピボットすることを試みます。これは、少なくとも 1 つのフィールドを使用してデータをグループ化し、少なくとも 1 つの集約を適用することを含みます。変換後のデータがどのように見えるかをプレビューできるので、ぜひ試してみてください！また、ヒストグラムチャートを有効にして、データ内の値の分布をよりよく理解することもできます。
    たとえば、データを製品 ID でグループ化し、各製品の総販売数と平均価格を計算したいかもしれません。あるいは、個々の顧客の行動を見て、各顧客が合計でどれだけ支出したか、どのくらいの異なる製品カテゴリを購入したかを計算したいかもしれません。また、通貨や地域を考慮に入れることもできます。このデータを変換し解釈する最も興味深い方法は何ですか？
    Kibana の Management
    > Stack Management
    > Data
    > Transforms に移動し、ウィザードを使用して変換を作成します:
    Kibana でのシンプルな変換の作成
    顧客 ID でデータをグループ化し、1 つ以上の集約を追加して各顧客の注文についてさらに学びます。たとえば、購入した製品の合計、購入の総価格、単一の注文で購入した最大製品数、および総注文数を計算してみましょう。これは、sum 集約 を total_quantity および taxless_total_price フィールドに、max 集約 を total_quantity フィールドに、cardinality 集約 を order_id フィールドに使用することで実現します:
    Kibana での変換に複数の集約を追加
    データのサブセットに興味がある場合は、オプションで query 要素を含めることができます。この例では、currency が EUR の注文のみを見ているようにデータをフィルタリングしました。あるいは、そのフィールドでもデータをグループ化できます。より複雑なクエリを使用したい場合は、saved search からデータフレームを作成できます。
    好ましい場合は、preview transforms API を使用できます。
    APIの例

Python

  1. resp = client.transform.preview_transform(
  2.    source={
  3.    "index": "kibana_sample_data_ecommerce",
  4.    "query": {
  5.    "bool": {
  6.    "filter": {
  7.    "term": {
  8.    "currency": "EUR"
  9.    }
  10.    }
  11.    }
  12.    }
  13.    },
  14.    pivot={
  15.    "group_by": {
  16.    "customer_id": {
  17.    "terms": {
  18.    "field": "customer_id"
  19.    }
  20.    }
  21.    },
  22.    "aggregations": {
  23.    "total_quantity.sum": {
  24.    "sum": {
  25.    "field": "total_quantity"
  26.    }
  27.    },
  28.    "taxless_total_price.sum": {
  29.    "sum": {
  30.    "field": "taxless_total_price"
  31.    }
  32.    },
  33.    "total_quantity.max": {
  34.    "max": {
  35.    "field": "total_quantity"
  36.    }
  37.    },
  38.    "order_id.cardinality": {
  39.    "cardinality": {
  40.    "field": "order_id"
  41.    }
  42.    }
  43.    }
  44.    },
  45.   )
  46.   print(resp)

Js

  1. const response = await client.transform.previewTransform({
  2.    source: {
  3.    index: "kibana_sample_data_ecommerce",
  4.    query: {
  5.    bool: {
  6.    filter: {
  7.    term: {
  8.    currency: "EUR",
  9.    },
  10.    },
  11.    },
  12.    },
  13.    },
  14.    pivot: {
  15.    group_by: {
  16.    customer_id: {
  17.    terms: {
  18.    field: "customer_id",
  19.    },
  20.    },
  21.    },
  22.    aggregations: {
  23.    "total_quantity.sum": {
  24.    sum: {
  25.    field: "total_quantity",
  26.    },
  27.    },
  28.    "taxless_total_price.sum": {
  29.    sum: {
  30.    field: "taxless_total_price",
  31.    },
  32.    },
  33.    "total_quantity.max": {
  34.    max: {
  35.    field: "total_quantity",
  36.    },
  37.    },
  38.    "order_id.cardinality": {
  39.    cardinality: {
  40.    field: "order_id",
  41.    },
  42.    },
  43.    },
  44.    },
  45.   });
  46.   console.log(response);

コンソール

  1. POST _transform/_preview
  2.   {
  3.    "source": {
  4.    "index": "kibana_sample_data_ecommerce",
  5.    "query": {
  6.    "bool": {
  7.    "filter": {
  8.    "term": {"currency": "EUR"}
  9.    }
  10.    }
  11.    }
  12.    },
  13.    "pivot": {
  14.    "group_by": {
  15.    "customer_id": {
  16.    "terms": {
  17.    "field": "customer_id"
  18.    }
  19.    }
  20.    },
  21.    "aggregations": {
  22.    "total_quantity.sum": {
  23.    "sum": {
  24.    "field": "total_quantity"
  25.    }
  26.    },
  27.    "taxless_total_price.sum": {
  28.    "sum": {
  29.    "field": "taxless_total_price"
  30.    }
  31.    },
  32.    "total_quantity.max": {
  33.    "max": {
  34.    "field": "total_quantity"
  35.    }
  36.    },
  37.    "order_id.cardinality": {
  38.    "cardinality": {
  39.    "field": "order_id"
  40.    }
  41.    }
  42.    }
  43.    }
  44.   }
  • 4. プレビューで表示される内容に満足したら、変換を作成します。
    • 4.1. 変換 ID、宛先インデックスの名前、およびオプションで説明を提供します。宛先インデックスが存在しない場合、変換を開始すると自動的に作成されます。
    • 4.2. 変換を一度だけ実行するか、継続的に実行するかを決定します。このサンプルデータインデックスは変更されないため、デフォルトの動作を使用して変換を一度だけ実行します。ただし、試してみたい場合は、Continuous mode をクリックしてください。変換が変更されたエンティティを確認するために使用できるフィールドを選択する必要があります。一般的には、インジェストタイムスタンプフィールドを使用するのが良いアイデアです。ただし、この例では order_date フィールドを使用できます。
    • 4.3. オプションで、変換に適用される保持ポリシーを構成できます。宛先インデックス内の古いドキュメントを特定するために使用される日付フィールドを選択し、最大年齢を提供します。設定された値よりも古いドキュメントは宛先インデックスから削除されます。
      Kibana での変換 ID と保持ポリシーの追加
      Kibana では、変換の作成を完了する前に、プレビュートランスフォーム API リクエストをクリップボードにコピーできます。この情報は、宛先インデックスを手動で作成するかどうかを決定する際に役立ちます。
      変換プレビューの Dev Console ステートメントをクリップボードにコピー
      好ましい場合は、create transforms API を使用できます。
      APIの例

Python

  1. resp = client.transform.put_transform(
  2.    transform_id="ecommerce-customer-transform",
  3.    source={
  4.    "index": [
  5.    "kibana_sample_data_ecommerce"
  6.    ],
  7.    "query": {
  8.    "bool": {
  9.    "filter": {
  10.    "term": {
  11.    "currency": "EUR"
  12.    }
  13.    }
  14.    }
  15.    }
  16.    },
  17.    pivot={
  18.    "group_by": {
  19.    "customer_id": {
  20.    "terms": {
  21.    "field": "customer_id"
  22.    }
  23.    }
  24.    },
  25.    "aggregations": {
  26.    "total_quantity.sum": {
  27.    "sum": {
  28.    "field": "total_quantity"
  29.    }
  30.    },
  31.    "taxless_total_price.sum": {
  32.    "sum": {
  33.    "field": "taxless_total_price"
  34.    }
  35.    },
  36.    "total_quantity.max": {
  37.    "max": {
  38.    "field": "total_quantity"
  39.    }
  40.    },
  41.    "order_id.cardinality": {
  42.    "cardinality": {
  43.    "field": "order_id"
  44.    }
  45.    }
  46.    }
  47.    },
  48.    dest={
  49.    "index": "ecommerce-customers"
  50.    },
  51.    retention_policy={
  52.    "time": {
  53.    "field": "order_date",
  54.    "max_age": "60d"
  55.    }
  56.    },
  57.   )
  58.   print(resp)

Js

  1. const response = await client.transform.putTransform({
  2.    transform_id: "ecommerce-customer-transform",
  3.    source: {
  4.    index: ["kibana_sample_data_ecommerce"],
  5.    query: {
  6.    bool: {
  7.    filter: {
  8.    term: {
  9.    currency: "EUR",
  10.    },
  11.    },
  12.    },
  13.    },
  14.    },
  15.    pivot: {
  16.    group_by: {
  17.    customer_id: {
  18.    terms: {
  19.    field: "customer_id",
  20.    },
  21.    },
  22.    },
  23.    aggregations: {
  24.    "total_quantity.sum": {
  25.    sum: {
  26.    field: "total_quantity",
  27.    },
  28.    },
  29.    "taxless_total_price.sum": {
  30.    sum: {
  31.    field: "taxless_total_price",
  32.    },
  33.    },
  34.    "total_quantity.max": {
  35.    max: {
  36.    field: "total_quantity",
  37.    },
  38.    },
  39.    "order_id.cardinality": {
  40.    cardinality: {
  41.    field: "order_id",
  42.    },
  43.    },
  44.    },
  45.    },
  46.    dest: {
  47.    index: "ecommerce-customers",
  48.    },
  49.    retention_policy: {
  50.    time: {
  51.    field: "order_date",
  52.    max_age: "60d",
  53.    },
  54.    },
  55.   });
  56.   console.log(response);

コンソール

  1. PUT _transform/ecommerce-customer-transform
  2.   {
  3.    "source": {
  4.    "index": [
  5.    "kibana_sample_data_ecommerce"
  6.    ],
  7.    "query": {
  8.    "bool": {
  9.    "filter": {
  10.    "term": {
  11.    "currency": "EUR"
  12.    }
  13.    }
  14.    }
  15.    }
  16.    },
  17.    "pivot": {
  18.    "group_by": {
  19.    "customer_id": {
  20.    "terms": {
  21.    "field": "customer_id"
  22.    }
  23.    }
  24.    },
  25.    "aggregations": {
  26.    "total_quantity.sum": {
  27.    "sum": {
  28.    "field": "total_quantity"
  29.    }
  30.    },
  31.    "taxless_total_price.sum": {
  32.    "sum": {
  33.    "field": "taxless_total_price"
  34.    }
  35.    },
  36.    "total_quantity.max": {
  37.    "max": {
  38.    "field": "total_quantity"
  39.    }
  40.    },
  41.    "order_id.cardinality": {
  42.    "cardinality": {
  43.    "field": "order_id"
  44.    }
  45.    }
  46.    }
  47.    },
  48.    "dest": {
  49.    "index": "ecommerce-customers"
  50.    },
  51.    "retention_policy": {
  52.    "time": {
  53.    "field": "order_date",
  54.    "max_age": "60d"
  55.    }
  56.    }
  57.   }
  • 5. オプション: 宛先インデックスを作成します。
    宛先インデックスが存在しない場合、変換を開始する最初の時に作成されます。ピボット変換は、ソースインデックスと変換集約から宛先インデックスのマッピングを推測します。宛先インデックスにスクリプトから派生したフィールドがある場合（たとえば、scripted_metrics または bucket_scripts 集約を使用する場合）、それらは dynamic mappings で作成されます。プレビュートランスフォーム API を使用して、宛先インデックスに使用されるマッピングをプレビューできます。Kibana では、API リクエストをクリップボードにコピーした場合、コンソールに貼り付けて、API レスポンス内の generated_dest_index オブジェクトを参照します。
    変換には、Kibana で利用可能なオプションよりも多くの構成オプションが API によって提供される場合があります。たとえば、Create transform を呼び出すことで dest のインジェストパイプラインを設定できます。すべての変換構成オプションについては、documentation を参照してください。
    APIの例

コンソール - 結果

  1. {
  2.    "preview" : [
  3.    {
  4.    "total_quantity" : {
  5.    "max" : 2,
  6.    "sum" : 118.0
  7.    },
  8.    "taxless_total_price" : {
  9.    "sum" : 3946.9765625
  10.    },
  11.    "customer_id" : "10",
  12.    "order_id" : {
  13.    "cardinality" : 59
  14.    }
  15.    },
  16.    ...
  17.    ],
  18.    "generated_dest_index" : {
  19.    "mappings" : {
  20.    "_meta" : {
  21.    "_transform" : {
  22.    "transform" : "transform-preview",
  23.    "version" : {
  24.    "created" : "8.0.0"
  25.    },
  26.    "creation_date_in_millis" : 1621991264061
  27.    },
  28.    "created_by" : "transform"
  29.    },
  30.    "properties" : {
  31.    "total_quantity.sum" : {
  32.    "type" : "double"
  33.    },
  34.    "total_quantity" : {
  35.    "type" : "object"
  36.    },
  37.    "taxless_total_price" : {
  38.    "type" : "object"
  39.    },
  40.    "taxless_total_price.sum" : {
  41.    "type" : "double"
  42.    },
  43.    "order_id.cardinality" : {
  44.    "type" : "long"
  45.    },
  46.    "customer_id" : {
  47.    "type" : "keyword"
  48.    },
  49.    "total_quantity.max" : {
  50.    "type" : "integer"
  51.    },
  52.    "order_id" : {
  53.    "type" : "object"
  54.    }
  55.    }
  56.    },
  57.    "settings" : {
  58.    "index" : {
  59.    "number_of_shards" : "1",
  60.    "auto_expand_replicas" : "0-1"
  61.    }
  62.    },
  63.    "aliases" : { }
  64.    }
  65.   }

場合によっては、推測されたマッピングが実際のデータと互換性がないことがあります。たとえば、数値オーバーフローが発生したり、動的にマッピングされたフィールドに数値と文字列の両方が含まれることがあります。この問題を回避するには、変換を開始する前に宛先インデックスを作成してください。詳細については、create index API を参照してください。
APIの例
変換プレビューからの情報を使用して宛先インデックスを作成できます。たとえば:

Python

  1. resp = client.indices.create(
  2.    index="ecommerce-customers",
  3.    mappings={
  4.    "properties": {
  5.    "total_quantity.sum": {
  6.    "type": "double"
  7.    },
  8.    "total_quantity": {
  9.    "type": "object"
  10.    },
  11.    "taxless_total_price": {
  12.    "type": "object"
  13.    },
  14.    "taxless_total_price.sum": {
  15.    "type": "double"
  16.    },
  17.    "order_id.cardinality": {
  18.    "type": "long"
  19.    },
  20.    "customer_id": {
  21.    "type": "keyword"
  22.    },
  23.    "total_quantity.max": {
  24.    "type": "integer"
  25.    },
  26.    "order_id": {
  27.    "type": "object"
  28.    }
  29.    }
  30.    },
  31.   )
  32.   print(resp)

Ruby

  1. response = client.indices.create(
  2.    index: 'ecommerce-customers',
  3.    body: {
  4.    mappings: {
  5.    properties: {
  6.    'total_quantity.sum' => {
  7.    type: 'double'
  8.    },
  9.    total_quantity: {
  10.    type: 'object'
  11.    },
  12.    taxless_total_price: {
  13.    type: 'object'
  14.    },
  15.    'taxless_total_price.sum' => {
  16.    type: 'double'
  17.    },
  18.    'order_id.cardinality' => {
  19.    type: 'long'
  20.    },
  21.    customer_id: {
  22.    type: 'keyword'
  23.    },
  24.    'total_quantity.max' => {
  25.    type: 'integer'
  26.    },
  27.    order_id: {
  28.    type: 'object'
  29.    }
  30.    }
  31.    }
  32.    }
  33.   )
  34.   puts response

Js

  1. const response = await client.indices.create({
  2.    index: "ecommerce-customers",
  3.    mappings: {
  4.    properties: {
  5.    "total_quantity.sum": {
  6.    type: "double",
  7.    },
  8.    total_quantity: {
  9.    type: "object",
  10.    },
  11.    taxless_total_price: {
  12.    type: "object",
  13.    },
  14.    "taxless_total_price.sum": {
  15.    type: "double",
  16.    },
  17.    "order_id.cardinality": {
  18.    type: "long",
  19.    },
  20.    customer_id: {
  21.    type: "keyword",
  22.    },
  23.    "total_quantity.max": {
  24.    type: "integer",
  25.    },
  26.    order_id: {
  27.    type: "object",
  28.    },
  29.    },
  30.    },
  31.   });
  32.   console.log(response);

コンソール

  1. PUT /ecommerce-customers
  2.   {
  3.    "mappings": {
  4.    "properties": {
  5.    "total_quantity.sum" : {
  6.    "type" : "double"
  7.    },
  8.    "total_quantity" : {
  9.    "type" : "object"
  10.    },
  11.    "taxless_total_price" : {
  12.    "type" : "object"
  13.    },
  14.    "taxless_total_price.sum" : {
  15.    "type" : "double"
  16.    },
  17.    "order_id.cardinality" : {
  18.    "type" : "long"
  19.    },
  20.    "customer_id" : {
  21.    "type" : "keyword"
  22.    },
  23.    "total_quantity.max" : {
  24.    "type" : "integer"
  25.    },
  26.    "order_id" : {
  27.    "type" : "object"
  28.    }
  29.    }
  30.    }
  31.   }
  • 6. 変換を開始します。
    リソースの利用状況はクラスターの負荷に基づいて自動的に調整されますが、変換が実行されている間、検索およびインデックスの負荷がクラスターに増加します。ただし、過剰な負荷が発生している場合は、停止できます。
    Kibana で変換を開始、停止、リセット、および管理できます:
    Kibana での変換の管理
    または、start transforms、stop transforms、および reset transforms API を使用できます。
    変換をリセットすると、すべてのチェックポイント、状態、および宛先インデックス（変換によって作成された場合）が削除されます。変換は、作成されたばかりのように再び開始する準備が整います。
    APIの例

Python

  1. resp = client.transform.start_transform(
  2.    transform_id="ecommerce-customer-transform",
  3.   )
  4.   print(resp)

Ruby

  1. response = client.transform.start_transform(
  2.    transform_id: 'ecommerce-customer-transform'
  3.   )
  4.   puts response

Js

  1. const response = await client.transform.startTransform({
  2.    transform_id: "ecommerce-customer-transform",
  3.   });
  4.   console.log(response);

コンソール

  1. POST _transform/ecommerce-customer-transform/_start

バッチ変換を選択した場合、それは単一の操作であり、単一のチェックポイントを持ちます。完了した後に再起動することはできません。継続的な変換は、新しいソースデータがインジェストされると、チェックポイントを継続的に増加させて処理します。

  • 7. 新しいインデックスのデータを探索します。
    たとえば、Kibana の Discover アプリケーションを使用します:
    Kibana での新しいインデックスの探索
  • 8. オプション: 今回は latest メソッドを使用して別の変換を作成します。
    このメソッドは、各ユニークキー値の最新のドキュメントで宛先インデックスを埋めます。たとえば、各顧客または各国および地域の最新の注文（order_date フィールドでソート）を見つけたいかもしれません。
    Kibana での最新の変換の作成
    APIの例

Python

  1. resp = client.transform.preview_transform(
  2.    source={
  3.    "index": "kibana_sample_data_ecommerce",
  4.    "query": {
  5.    "bool": {
  6.    "filter": {
  7.    "term": {
  8.    "currency": "EUR"
  9.    }
  10.    }
  11.    }
  12.    }
  13.    },
  14.    latest={
  15.    "unique_key": [
  16.    "geoip.country_iso_code",
  17.    "geoip.region_name"
  18.    ],
  19.    "sort": "order_date"
  20.    },
  21.   )
  22.   print(resp)

Js

  1. const response = await client.transform.previewTransform({
  2.    source: {
  3.    index: "kibana_sample_data_ecommerce",
  4.    query: {
  5.    bool: {
  6.    filter: {
  7.    term: {
  8.    currency: "EUR",
  9.    },
  10.    },
  11.    },
  12.    },
  13.    },
  14.    latest: {
  15.    unique_key: ["geoip.country_iso_code", "geoip.region_name"],
  16.    sort: "order_date",
  17.    },
  18.   });
  19.   console.log(response);

コンソール

  1. POST _transform/_preview
  2.   {
  3.    "source": {
  4.    "index": "kibana_sample_data_ecommerce",
  5.    "query": {
  6.    "bool": {
  7.    "filter": {
  8.    "term": {"currency": "EUR"}
  9.    }
  10.    }
  11.    }
  12.    },
  13.    "latest": {
  14.    "unique_key": ["geoip.country_iso_code", "geoip.region_name"],
  15.    "sort": "order_date"
  16.    }
  17.   }

宛先インデックスが存在しない場合、変換を開始する最初の時に作成されます。ただし、最新の変換は、インデックスを作成する際にマッピング定義を推測しません。代わりに、動的マッピングを使用します。明示的なマッピングを使用するには、変換を開始する前に宛先インデックスを作成してください。

  • 9. 変換を保持したくない場合は、Kibana で削除するか、delete transform API を使用できます。デフォルトでは、変換を削除すると、その宛先インデックスと Kibana インデックスパターンは残ります。

Kibana サンプルデータのシンプルな変換を作成したので、自分のデータに対する可能なユースケースを考えてみてください。さらにアイデアが必要な場合は、When to use transforms および Examples を参照してください。