ロールアップの始め方

8.11.0で非推奨。

ロールアップは将来のバージョンで削除されます。代わりにmigrateをダウンサンプリングに移行してください。

8.15.0以降、ロールアップを使用していないクラスターでput job APIを呼び出すと、ロールアップの非推奨と計画的な削除に関するメッセージと共に失敗します。クラスターは、put job APIを実行するためにロールアップジョブまたはロールアップインデックスを含む必要があります。

ロールアップ機能を使用するには、1つ以上の「ロールアップジョブ」を作成する必要があります。これらのジョブはバックグラウンドで継続的に実行され、指定したインデックスまたはインデックスをロールアップし、ロールアップされたドキュメントを選択した二次インデックスに配置します。

センサーデータを保持する一連の日次インデックス（sensor-2017-01-01、sensor-2017-01-02など）があると想像してください。サンプルドキュメントは次のようになります:

Js

  1. {
  2.   "timestamp": 1516729294000,
  3.   "temperature": 200,
  4.   "voltage": 5.2,
  5.   "node": "a"
  6. }

ロールアップジョブの作成

これらのドキュメントを時間ごとの要約にロールアップしたいと考えています。これにより、1時間以上の任意の時間間隔でレポートやダッシュボードを生成できます。ロールアップジョブは次のようになります:

Python

  1. resp = client.rollup.put_job(
  2.    id="sensor",
  3.    index_pattern="sensor-*",
  4.    rollup_index="sensor_rollup",
  5.    cron="*/30 * * * * ?",
  6.    page_size=1000,
  7.    groups={
  8.    "date_histogram": {
  9.    "field": "timestamp",
  10.    "fixed_interval": "60m"
  11.    },
  12.    "terms": {
  13.    "fields": [
  14.    "node"
  15.    ]
  16.    }
  17.    },
  18.    metrics=[
  19.    {
  20.    "field": "temperature",
  21.    "metrics": [
  22.    "min",
  23.    "max",
  24.    "sum"
  25.    ]
  26.    },
  27.    {
  28.    "field": "voltage",
  29.    "metrics": [
  30.    "avg"
  31.    ]
  32.    }
  33.    ],
  34. )
  35. print(resp)

Js

  1. const response = await client.rollup.putJob({
  2.   id: "sensor",
  3.   index_pattern: "sensor-*",
  4.   rollup_index: "sensor_rollup",
  5.   cron: "*/30 * * * * ?",
  6.   page_size: 1000,
  7.   groups: {
  8.    date_histogram: {
  9.    field: "timestamp",
  10.    fixed_interval: "60m",
  11.    },
  12.    terms: {
  13.    fields: ["node"],
  14.    },
  15.   },
  16.   metrics: [
  17.    {
  18.    field: "temperature",
  19.    metrics: ["min", "max", "sum"],
  20.    },
  21.    {
  22.    field: "voltage",
  23.    metrics: ["avg"],
  24.    },
  25.   ],
  26. });
  27. console.log(response);

コンソール

  1. PUT _rollup/job/sensor
  2. {
  3.   "index_pattern": "sensor-*",
  4.   "rollup_index": "sensor_rollup",
  5.   "cron": "*/30 * * * * ?",
  6.   "page_size": 1000,
  7.   "groups": {
  8.    "date_histogram": {
  9.    "field": "timestamp",
  10.    "fixed_interval": "60m"
  11.    },
  12.    "terms": {
  13.    "fields": [ "node" ]
  14.    }
  15.   },
  16.   "metrics": [
  17.    {
  18.    "field": "temperature",
  19.    "metrics": [ "min", "max", "sum" ]
  20.    },
  21.    {
  22.    "field": "voltage",
  23.    "metrics": [ "avg" ]
  24.    }
  25.   ]
  26. }

ジョブに「sensor」というIDを与え（URL内：PUT _rollup/job/sensor）、インデックスパターン"sensor-*"をロールアップするように指示します。このジョブは、そのパターンに一致するインデックスを見つけてロールアップします。ロールアップの要約は"sensor_rollup"インデックスに保存されます。 

  2. 代わりに、cronが毎日真夜中に1回実行されるように設定されている場合、ジョブは過去24時間分のデータを処理します。この選択は、ロールアップの「リアルタイム」性や、継続的に処理するかオフピーク時間に移動するかに基づいて、主に好みによります。  
  4. 次に、`````groups`````のセットを定義します。基本的に、データをクエリする際に後でピボットしたい次元を定義しています。このジョブのグループ化により、`````date_histogram`````集計を`````timestamp`````フィールドに対して、時間ごとの間隔でロールアップできます。また、`````node`````フィールドに対して用語集計を実行することもできます。  
  6. **日付ヒストグラムの間隔とcronスケジュール**  
  8. ジョブのcronが30秒ごとに実行されるように設定されている一方で、date_histogramは60分間隔でロールアップするように設定されていることに注意してください。これらはどのように関連していますか？  
  10. date_histogramは保存されたデータの粒度を制御します。データは時間ごとの間隔でロールアップされ、より細かい粒度でクエリすることはできません。cronは単にプロセスがロールアップする新しいデータを探すタイミングを制御します。30秒ごとに新しい1時間分のデータがあるかどうかを確認し、ロールアップします。なければ、ジョブは再びスリープに戻ります。  
  12. 多くの場合、大きな間隔（1時間）に対して非常に小さなcron（30秒）を定義することは理にかなわないことが多いです。なぜなら、ほとんどのアクティベーションは単にスリープに戻るからです。しかし、それでも問題はありません。ジョブは正しいことを行います。  
  14. データのために生成されるべきグループを定義した後、次に収集すべきメトリックを構成します。デフォルトでは、各グループに対して`````doc_counts`````のみが収集されます。ロールアップを有用にするために、平均、最小、最大などのメトリックを追加することがよくあります。この例では、メトリックは非常に単純です：`````temperature`````フィールドの最小/最大/合計を保存し、`````voltage`````フィールドの平均を保存したいと考えています。  
  16. **平均は合成できない？！**  
  18. 以前にロールアップを使用したことがある場合、平均に対して慎重になるかもしれません。10分間隔で保存された平均は、通常は大きな間隔には役に立ちません。6つの10分間の平均を平均して1時間の平均を求めることはできません。平均の平均は総平均に等しくありません。  
  20. このため、他のシステムは平均を省略するか、より柔軟なクエリをサポートするために複数の間隔で平均を保存する傾向があります。  
  22. 代わりに、データロールアップ機能は、定義された時間間隔のために`````count`````と`````sum`````を保存します。これにより、定義された間隔以上の任意の間隔で平均を再構築できます。これにより、最小限のストレージコストで最大の柔軟性が得られます…そして、平均の正確性について心配する必要はありません（ここでは平均の平均はありません！）  
  24. ジョブ構文の詳細については、[ロールアップジョブの作成](/read/elasticsearch-8-15/72462e080b27c599.md)を参照してください。  
  26. 上記のコマンドを実行してジョブを作成すると、次のような応答が得られます：  
  27.  [](#bc5fcc40c29087a0df7b5405bb70de5c)
  30. #### コンソール-結果
  32. ``````console-result
  33. {
  34.   "acknowledged": true
  35. }
  36. `

ジョブの開始

ジョブが作成されると、非アクティブ状態になります。ジョブはデータの処理を開始する前に開始する必要があります（これにより、後で一時的に停止するために構成を削除せずに停止できます）。

ジョブを開始するには、このコマンドを実行します：

Python

  1. resp = client.rollup.start_job(
  2.    id="sensor",
  3. )
  4. print(resp)

Ruby

  1. response = client.rollup.start_job(
  2.   id: 'sensor'
  3. )
  4. puts response

Js

  1. const response = await client.rollup.startJob({
  2.   id: "sensor",
  3. });
  4. console.log(response);

コンソール

  1. POST _rollup/job/sensor/_start

ロールアップ結果の検索

ジョブが実行され、いくつかのデータが処理された後、ロールアップ検索エンドポイントを使用して検索を行うことができます。ロールアップ機能は、慣れ親しんだ同じQuery DSL構文を使用できるように設計されています…ただし、ロールアップされたデータで実行されます。

たとえば、このクエリを考えてみてください：

Python

  1. resp = client.rollup.rollup_search(
  2.    index="sensor_rollup",
  3.    size=0,
  4.    aggregations={
  5.    "max_temperature": {
  6.    "max": {
  7.    "field": "temperature"
  8.    }
  9.    }
  10.    },
  11. )
  12. print(resp)

Ruby

  1. response = client.rollup.rollup_search(
  2.   index: 'sensor_rollup',
  3.   body: {
  4.    size: 0,
  5.    aggregations: {
  6.    max_temperature: {
  7.    max: {
  8.    field: 'temperature'
  9.    }
  10.    }
  11.    }
  12.   }
  13. )
  14. puts response

Js

  1. const response = await client.rollup.rollupSearch({
  2.   index: "sensor_rollup",
  3.   size: 0,
  4.   aggregations: {
  5.    max_temperature: {
  6.    max: {
  7.    field: "temperature",
  8.    },
  9.    },
  10.   },
  11. });
  12. console.log(response);

コンソール

  1. GET /sensor_rollup/_rollup_search
  2. {
  3.   "size": 0,
  4.   "aggregations": {
  5.    "max_temperature": {
  6.    "max": {
  7.    "field": "temperature"
  8.    }
  9.    }
  10.   }
  11. }

これは、temperatureフィールドの最大値を計算する単純な集計です。しかし、sensor_rollupインデックスに送信されていることに気付くでしょう。生のsensor-*インデックスではなく、_rollup_searchエンドポイントを使用していることにも気付くでしょう。そうでなければ、構文は期待通りです。

そのクエリを実行すると、通常の集計応答のように見える結果が得られます：

コンソール-結果

  1. {
  2.   "took" : 102,
  3.   "timed_out" : false,
  4.   "terminated_early" : false,
  5.   "_shards" : ... ,
  6.   "hits" : {
  7.    "total" : {
  8.    "value": 0,
  9.    "relation": "eq"
  10.    },
  11.    "max_score" : 0.0,
  12.    "hits" : [ ]
  13.   },
  14.   "aggregations" : {
  15.    "max_temperature" : {
  16.    "value" : 202.0
  17.    }
  18.   }
  19. }

唯一の顕著な違いは、ロールアップ検索結果にはhitsがゼロであることです。なぜなら、もはや元のライブデータを検索していないからです。それ以外は、構文は同一です。

ここにはいくつかの興味深いポイントがあります。まず、データは時間ごとの間隔でロールアップされ、ノード名で分割されているにもかかわらず、実行したクエリはすべてのドキュメントの最大温度を計算しています。ジョブで構成されたgroupsはクエリの必須要素ではなく、単にパーティションを分けるための追加の次元です。第二に、リクエストと応答の構文は通常のDSLとほぼ同じであり、ダッシュボードやアプリケーションに統合しやすくなっています。

最後に、定義したグループフィールドを使用して、より複雑なクエリを構築できます：

Python

  1. resp = client.rollup.rollup_search(
  2.    index="sensor_rollup",
  3.    size=0,
  4.    aggregations={
  5.    "timeline": {
  6.    "date_histogram": {
  7.    "field": "timestamp",
  8.    "fixed_interval": "7d"
  9.    },
  10.    "aggs": {
  11.    "nodes": {
  12.    "terms": {
  13.    "field": "node"
  14.    },
  15.    "aggs": {
  16.    "max_temperature": {
  17.    "max": {
  18.    "field": "temperature"
  19.    }
  20.    },
  21.    "avg_voltage": {
  22.    "avg": {
  23.    "field": "voltage"
  24.    }
  25.    }
  26.    }
  27.    }
  28.    }
  29.    }
  30.    },
  31. )
  32. print(resp)

Ruby

  1. response = client.rollup.rollup_search(
  2.   index: 'sensor_rollup',
  3.   body: {
  4.    size: 0,
  5.    aggregations: {
  6.    timeline: {
  7.    date_histogram: {
  8.    field: 'timestamp',
  9.    fixed_interval: '7d'
  10.    },
  11.    aggregations: {
  12.    nodes: {
  13.    terms: {
  14.    field: 'node'
  15.    },
  16.    aggregations: {
  17.    max_temperature: {
  18.    max: {
  19.    field: 'temperature'
  20.    }
  21.    },
  22.    avg_voltage: {
  23.    avg: {
  24.    field: 'voltage'
  25.    }
  26.    }
  27.    }
  28.    }
  29.    }
  30.    }
  31.    }
  32.   }
  33. )
  34. puts response

Js

  1. const response = await client.rollup.rollupSearch({
  2.   index: "sensor_rollup",
  3.   size: 0,
  4.   aggregations: {
  5.    timeline: {
  6.    date_histogram: {
  7.    field: "timestamp",
  8.    fixed_interval: "7d",
  9.    },
  10.    aggs: {
  11.    nodes: {
  12.    terms: {
  13.    field: "node",
  14.    },
  15.    aggs: {
  16.    max_temperature: {
  17.    max: {
  18.    field: "temperature",
  19.    },
  20.    },
  21.    avg_voltage: {
  22.    avg: {
  23.    field: "voltage",
  24.    },
  25.    },
  26.    },
  27.    },
  28.    },
  29.    },
  30.   },
  31. });
  32. console.log(response);

コンソール

  1. GET /sensor_rollup/_rollup_search
  2. {
  3.   "size": 0,
  4.   "aggregations": {
  5.    "timeline": {
  6.    "date_histogram": {
  7.    "field": "timestamp",
  8.    "fixed_interval": "7d"
  9.    },
  10.    "aggs": {
  11.    "nodes": {
  12.    "terms": {
  13.    "field": "node"
  14.    },
  15.    "aggs": {
  16.    "max_temperature": {
  17.    "max": {
  18.    "field": "temperature"
  19.    }
  20.    },
  21.    "avg_voltage": {
  22.    "avg": {
  23.    "field": "voltage"
  24.    }
  25.    }
  26.    }
  27.    }
  28.    }
  29.    }
  30.   }
  31. }

それに対応する応答が返されます：

コンソール-結果

  1. {
  2.    "took" : 93,
  3.    "timed_out" : false,
  4.    "terminated_early" : false,
  5.    "_shards" : ... ,
  6.    "hits" : {
  7.    "total" : {
  8.    "value": 0,
  9.    "relation": "eq"
  10.    },
  11.    "max_score" : 0.0,
  12.    "hits" : [ ]
  13.    },
  14.    "aggregations" : {
  15.    "timeline" : {
  16.    "buckets" : [
  17.    {
  18.    "key_as_string" : "2018-01-18T00:00:00.000Z",
  19.    "key" : 1516233600000,
  20.    "doc_count" : 6,
  21.    "nodes" : {
  22.    "doc_count_error_upper_bound" : 0,
  23.    "sum_other_doc_count" : 0,
  24.    "buckets" : [
  25.    {
  26.    "key" : "a",
  27.    "doc_count" : 2,
  28.    "max_temperature" : {
  29.    "value" : 202.0
  30.    },
  31.    "avg_voltage" : {
  32.    "value" : 5.1499998569488525
  33.    }
  34.    },
  35.    {
  36.    "key" : "b",
  37.    "doc_count" : 2,
  38.    "max_temperature" : {
  39.    "value" : 201.0
  40.    },
  41.    "avg_voltage" : {
  42.    "value" : 5.700000047683716
  43.    }
  44.    },
  45.    {
  46.    "key" : "c",
  47.    "doc_count" : 2,
  48.    "max_temperature" : {
  49.    "value" : 202.0
  50.    },
  51.    "avg_voltage" : {
  52.    "value" : 4.099999904632568
  53.    }
  54.    }
  55.    ]
  56.    }
  57.    }
  58.    ]
  59.    }
  60.    }
  61. }

より複雑（日時ヒストグラムと用語集計、さらに追加の平均メトリック）であることに加えて、date_histogramは7d間隔を使用していることに気付くでしょう。60mの代わりに。

結論

このクイックスタートは、ロールアップが提供するコア機能の簡潔な概要を提供したはずです。ロールアップを設定する際に考慮すべきヒントや事項がさらにあります。これらはこのセクションの残りの部分で見つけることができます。また、利用可能なものの概要についてはREST APIを探索することもできます。