Elasticsearch:aliyun-timestream プラグインを使用した Elasticsearch 時系列データ管理の開始

aliyun-timestream プラグインでサポートされているインデックス作成 API を呼び出して、test_stream という名前の時系列インデックスを作成します。

PUT _time_stream/test_stream

次のコマンドを実行して、test_stream インデックスの構成を表示できます。

GET _time_stream/test_stream

コマンドが正常に実行されると、次の結果が返されます。

{
  "time_streams" : {
    "test_stream" : {
      "name" : "test_stream",
      "data_stream_name" : "test_stream",
      "time_stream_config" : {
        "labels_fields" : {
          "includes" : [
            "labels.*"
          ],
          "excludes" : [ ]
        },
        "metrics_fields" : {
          "includes" : [
            "metrics.*"
          ],
          "excludes" : [ ]
        },
        "label_prefix" : "labels.",
        "metric_prefix" : "metrics.",
        "downsample" : [ ]
      },
      "template_name" : ".timestream_test_stream",
      "template" : {
        "index_patterns" : [
          "test_stream"
        ],
        "template" : {
          "settings" : {
            "index" : {
              "mode" : "time_series",
              "codec" : "ali",
              "refresh_interval" : "10s",
              "ali_codec_service" : {
                "enabled" : "true",
                "source_reuse_doc_values" : {
                  "enabled" : "true"
                }
              },
              "translog" : {
                "durability" : "ASYNC"
              },
              "doc_value" : {
                "compression" : {
                  "default" : "zstd"
                }
              },
              "postings" : {
                "compression" : "zstd"
              },
              "source" : {
                "compression" : "zstd"
              },
              "routing_path" : [
                "labels.*"
              ]
            }
          },
          "mappings" : {
            "numeric_detection" : true,
            "dynamic_templates" : [
              {
                "labels_template_match_labels.*" : {
                  "path_match" : "labels.*",
                  "mapping" : {
                    "time_series_dimension" : "true",
                    "type" : "keyword"
                  },
                  "match_mapping_type" : "*"
                }
              },
              {
                "metrics_double_match_metrics.*" : {
                  "path_match" : "metrics.*",
                  "mapping" : {
                    "index" : "false",
                    "type" : "double"
                  },
                  "match_mapping_type" : "double"
                }
              },
              {
                "metrics_long_match_metrics.*" : {
                  "path_match" : "metrics.*",
                  "mapping" : {
                    "index" : "false",
                    "type" : "long"
                  },
                  "match_mapping_type" : "long"
                }
              }
            ],
            "properties" : {
              "@timestamp" : {
                "format" : "epoch_millis||strict_date_optional_time",
                "type" : "date"
              }
            }
          }
        },
        "composed_of" : [ ],
        "data_stream" : {
          "hidden" : false
        }
      },
      "version" : 1
    }
  }
}

返された結果は、インデックス作成 API によって インデックステンプレート.timestream_test_stream という名前の

パラメーター	説明
index.mode	値 time_series は、インデックスタイプが time_series であることを示します。システムは、時系列シナリオにおける Elasticsearch ベストプラクティスの構成をインデックスに自動的に統合します。
index.codec	値 ali は、aliyun-codec インデックス圧縮プラグイン が使用されることを示します。このパラメーターは、以下のパラメーターと共に使用して、占有されるディスク容量を削減できます。 index.ali_codec_service.enabled=true: aliyun-codec プラグインのインデックス圧縮機能が有効になっていることを示します。 index.doc_value.compression.default=zstd: zstd 圧縮アルゴリズムを使用して、列指向ドキュメントを圧縮することを示します。 index.postings.compression=zstd: zstd 圧縮アルゴリズムを使用して、転置インデックスドキュメントを圧縮することを示します。 index.ali_codec_service.source_reuse_doc_values.enabled=true: source_reuse_doc_values 機能が有効になっていることを示します。 index.source.compression=zstd: zstd 圧縮アルゴリズムを使用して、行指向ドキュメントを圧縮することを示します。

が作成されたことを示しています。 次の表は、テンプレートの設定で構成されているパラメータについて説明しています。

インデックステンプレートのマッピングには、時系列モデル用の dynamic_templates パラメータが含まれています。

• ディメンションフィールド: デフォルトでは、keyword タイプが使用されます。 ディメンションフィールドの time_series_dimension の値は true です。 index.mode の値が time_series の場合、time_series_dimension が true であるすべてのフィールドは、_tsid という名前の内部タイムラインフィールドに結合されます。
• メトリックフィールド: double タイプと long タイプがサポートされています。 メトリックフィールドは doc 値のみを保存し、インデックスデータは保存しません。

インデックステンプレートをカスタマイズする場合は、インデックステンプレートの構文を使用できます。 次のコードは例を示しています。

• インデックスのプライマリシャード数をカスタマイズする

  PUT _time_stream/test_stream
  {
    "template": {
      "settings": {
        "index": {
          "number_of_shards": "2"  // プライマリシャード数を 2 に設定
        }
      }
    }
  }
                          

• インデックスのデータモデルをカスタマイズする

  PUT _time_stream/test_stream
  {
    "template": {
      "settings": {
        "index": {
          "number_of_shards": "2" // プライマリシャード数を 2 に設定
        }
      }
    },
    "time_stream": {
      "labels_fields": ["labels_*"], // ラベルフィールドのプレフィックスを指定
      "metrics_fields": ["metrics_*"] // メトリックフィールドのプレフィックスを指定
  }

test_stream インデックスのプライマリシャード数を更新するには、次のコマンドを実行します。

POST _time_stream/test_stream/_update
{
  "template": {
    "settings": {
      "index": {
        "number_of_shards": "4" // プライマリシャード数を 4 に更新
      }
    }
  }
}

test_stream インデックスを削除するには、次のコマンドを実行します。

Delete _time_stream/test_stream

bulk API または index API を呼び出して、時系列モデルに基づいてデータを書き込みます。 次のコードは例を示しています。 @timestamp は現在の時刻を示す値に設定する必要があります。

POST test_stream/_doc
{
  "@timestamp": 1630465208722, // タイムスタンプ
  "metrics": { // メトリック
    "cpu.idle": 79.67298116109929,
    "disk_ioutil": 17.630910821570456,
    "mem.free": 75.79973639970004
  },
  "labels": { // ラベル（ディメンション）
    "disk_type": "disk_type2",
    "namespace": "namespaces1",
    "clusterId": "clusterId3",
    "nodeId": "nodeId5"
  }
}

時系列インデックスの時間範囲は、データストリーム機能に基づいて aliyun-timestream プラグインによって生成されます。 データストリーム内の各インデックスには、index.time_series.start_time パラメータと index.time_series.end_time パラメータが構成されています。 書き込むデータの @timestamp で示される時刻がインデックスの [start_time,end_time] 時間範囲内にある場合、データはインデックスに書き込むことができます。 時間範囲は、インデックスのデータストリーム機能によって生成されるため、時間範囲を構成する必要はありません。 インデックスが作成されると、インデックスの index.time_series.start_time の値は固定され、インデックスの index.time_series.end_time の値は、インデックスがロールオーバーされて新しいインデックスが生成されるまで増加し続けます。 この場合、元のインデックスの index.time_series.end_time の値は固定され、新しいインデックスの index.time_series.start_time の値は元のインデックスの index.time_series.end_time の値と等しくなります。 このように、データストリームには連続した時間範囲のインデックスが含まれています。

_search API を呼び出してデータをクエリします。

GET test_stream/_search

_cat/indices API を呼び出して、インデックスの詳細を表示します。

GET _cat/indices/test_stream?v&s=i

_stats API を呼び出して、インデックスの統計情報を表示します。

GET _time_stream/test_stream/_stats

コマンドが正常に実行されると、次の結果が返されます。

{
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "failed" : 0
  },
  "time_stream_count" : 1,
  "indices_count" : 1,
  "total_store_size_bytes" : 19132,
  "time_streams" : [
    {
      "time_stream" : "test_stream",
      "indices_count" : 1,
      "store_size_bytes" : 19132,
      "tsid_count" : 2
    }
  ]
}

Prometheus API を呼び出してデータをクエリする前に、Prometheus データソースを構成する必要があります。 次のいずれかの方法を使用して、データソースを構成できます。

• Grafana コンソールでデータソースを構成する
  Grafana コンソールで Prometheus データソースを構成し、URL に /_time_stream/prom/test_stream URI を指定して、時系列インデックスを Grafana の Prometheus データソースとして直接使用します（次の図を参照）。
• Prometheus API を呼び出してデータソースを構成する

  Prometheus API を呼び出して、メトリックフィールドとディメンションフィールドのキーワードのプレフィックスとサフィックスを削除します。 Prometheus API を使用してデフォルトの時系列データモデルに基づいて書き込まれたデータをクエリする場合、返されるメトリックフィールドのプレフィックス metrics. は表示されず、返されるディメンションフィールドのプレフィックス labels. は表示されません。

  時系列インデックスを作成するときにデータモデルをカスタマイズする場合は、メトリックフィールドとディメンションフィールドのプレフィックスとサフィックスを指定する必要があります。 そうしないと、Prometheus API を使用して時系列インデックス内のデータをクエリすると、実際のデータが返されます。 次のコードは、プレフィックスとサフィックスを指定する方法の例を示しています。

  PUT _time_stream/{name} // 時系列インデックス名
  {
    "time_stream": {
      "labels_fields": "@labels.*_l", // ラベルフィールドのプレフィックスとサフィックスを指定
      "metrics_fields": "@metrics.*_m", // メトリックフィールドのプレフィックスとサフィックスを指定
      "label_prefix": "@labels.", // ラベルのプレフィックス
      "label_suffix": "_l", // ラベルのサフィックス
      "metric_prefix": "@metrics.", // メトリックのプレフィックス
      "metric_suffix": "_m" // メトリックのサフィックス
    }
  }

Prometheus インスタントクエリ API と範囲クエリ API を使用して、Elasticsearch データに対して PromQL クエリを実行できます。 PromQL のサポートの詳細については、「aliyun-timestream による PromQL のサポート」をご参照ください。

• Prometheus インスタントクエリ API を呼び出してデータをクエリする

  GET /_time_stream/prom/test_stream/query?query=cpu.idle&time=1655769837

  説明 time の単位は秒です。 このパラメータを構成しない場合、デフォルトでは過去 5 分以内のデータがクエリされます。
  コマンドが正常に実行されると、次の結果が返されます。

  {
    "status" : "success",
    "data" : {
      "resultType" : "vector",
      "result" : [
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId1",
            "disk_type" : "disk_type1",
            "namespace" : "namespaces2",
            "nodeId" : "nodeId2"
          },
          "value" : [
            1655769837,
            "79.672981161"
          ]
        },
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId1",
            "disk_type" : "disk_type1",
            "namespace" : "namespaces2",
            "nodeId" : "nodeId5"
          },
          "value" : [
            1655769837,
            "79.672981161"
          ]
        },
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId3",
            "disk_type" : "disk_type2",
            "namespace" : "namespaces1",
            "nodeId" : "nodeId5"
          },
          "value" : [
            1655769837,
            "79.672981161"
          ]
        }
      ]
    }
  }

• Prometheus 範囲クエリ API を呼び出してデータをクエリする

  GET /_time_stream/prom/test_stream/query_range?query=cpu.idle&start=1655769800&end=16557699860&step=1m

  コマンドが正常に実行されると、次の結果が返されます。

  {
    "status" : "success",
    "data" : {
      "resultType" : "matrix",
      "result" : [
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId1",
            "disk_type" : "disk_type1",
            "namespace" : "namespaces2",
            "nodeId" : "nodeId2"
          },
          "value" : [
            [
              1655769860,
              "79.672981161"
            ]
          ]
        },
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId1",
            "disk_type" : "disk_type1",
            "namespace" : "namespaces2",
            "nodeId" : "nodeId5"
          },
          "value" : [
            [
              1655769860,
              "79.672981161"
            ]
          ]
        },
        {
          "metric" : {
            "__name__" : "cpu.idle",
            "clusterId" : "clusterId3",
            "disk_type" : "disk_type2",
            "namespace" : "namespaces1",
            "nodeId" : "nodeId5"
          },
          "value" : [
            [
              1655769860,
              "79.672981161"
            ]
          ]
        }
      ]
    }
  }

ダウンサンプリングは、時系列シナリオで大規模なデータクエリの高速化によく使用される機能です。 この機能の詳細については、「ダウンサンプリングの使用上の注意」をご参照ください。 aliyun-timestream プラグインでサポートされている API を使用して時系列インデックスを作成するときに、インデックスのダウンサンプリングルールを構成できます。 次のコードは例を示しています。

PUT _time_stream/test_stream
{
  "time_stream": {
    "downsample": [ // ダウンサンプリングルール
      {
        "interval": "1m" // 1 分間隔
      },
      {
        "interval": "10m" // 10 分間隔
      },
      {
        "interval": "60m" // 60 分間隔
      }
    ]
  }
}

ダウンサンプリングインデックスの生成プロセスを以下に示します。

1. 元のインデックスに対してダウンサンプリングを実行して、ダウンサンプリングインデックスを生成します。 元のインデックスがロールオーバーされると、新しいインデックスが生成され、特定の期間内に元のインデックスにデータは書き込まれません。 その後、元のインデックスに対してダウンサンプリングが実行されます。 デフォルトでは、現在の時刻が元のインデックスの end_time で示される時刻より 2 時間後であれば、ダウンサンプリングが開始されます。 この効果をシミュレートするために、インデックスを作成するときに start_time と end_time を手動で構成できます。
   重要 システムは、新しいインデックスの end_time の値を最新の時間に更新します。これはダウンサンプリングのデモに影響します。 デフォルトでは、値は 5 分間隔で変更されます。 ダウンサンプリングのデモでは、end_time の値が変更されていないことを確認する必要があります。 終了時刻GET {index}/_settings コマンドを実行して、 の値を表示できます。

   PUT _time_stream/test_stream
   {
     "template": {
       "settings": {
         "index.time_series.start_time": "2022-06-20T00:00:00.000Z", // 開始時刻
         "index.time_series.end_time": "2022-06-21T00:00:00.000Z" // 終了時刻
       }
     },
     "time_stream": {
       "downsample": [
         {
           "interval": "1m"
         },
         {
           "interval": "10m"
         },
         {
           "interval": "60m"
         }
       ]
     }
   }

2. インデックスの end_time を現在の時刻より 2 時間以上前の時点に設定し、インデックスにデータを書き込みます。 また、@timestamp を start_time と end_time の間の時刻を示す値に設定します。

   POST test_stream/_doc
   {
     "@timestamp": 1655706106000, // タイムスタンプ
     "metrics": {
       "cpu.idle": 79.67298116109929,
       "disk_ioutil": 17.630910821570456,
       "mem.free": 75.79973639970004
     },
     "labels": {
       "disk_type": "disk_type2",
       "namespace": "namespaces1",
       "clusterId": "clusterId3",
       "nodeId": "nodeId5"
     }
   }

3. インデックスにデータが書き込まれたら、インデックスから start_time と end_time を削除します。

   POST _time_stream/test_stream/_update
   {
     "time_stream": {
       "downsample": [
         {
           "interval": "1m"
         },
         {
           "interval": "10m"
         },
         {
           "interval": "60m"
         }
       ]
     }
   }

4. インデックスをロールオーバーします。

   POST test_stream/_rollover

5. ロールオーバーが完了したら、GET _cat/indices/test_stream?v&s=i コマンドを実行して、test_stream インデックスによって生成されたダウンサンプリングインデックスを表示します。
   コマンドが正常に実行されると、次の結果が返されます。

   health status index                                          uuid                   pri rep docs.count docs.deleted store.size pri.store.size
   green  open   .ds-test_stream-2022.06.21-000001              vhEwKIlwSGO3ax4RKn****   1   1          9            0     18.5kb         12.1kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_10m r9Tsj0v-SyWJDc64oC****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_1h  cKsAlMK-T2-luefNAF****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000001_interval_1m  L6ocasDFTz-c89KjND****   1   1          1            0     15.8kb          7.9kb
   green  open   .ds-test_stream-2022.06.21-000002              42vlHEFFQrmMAdNdCz****   1   1          0            0       452b           226b

ダウンサンプリングインデックス内のデータをクエリするときは、クエリするデータの時間範囲に基づいて適切な interval を指定できます。 次の図に示すように、元のインデックスの名前（test_stream など）が渡され、interval パラメータは date_histogram パラメータを使用して渡されます。 システムがダウンサンプリングを実行すると、データクエリに最適な時間精度を持つインデックスを自動的に選択できます。

たとえば、次のクエリを実行します。

GET test_stream/_search?size=0&request_cache=false
{
  "aggs": { // 集計
    "1": { // 集計 1
      "terms": { // ラベルの disk_type フィールドでグループ化
        "field": "labels.disk_type",
        "size": 10
      },
      "aggs": { // 集計 2
        "2": {
          "date_histogram": { // タイムスタンプでヒストグラムを作成
            "field": "@timestamp",
            "fixed_interval": "120m" // 120 分間隔
          }
        }
      }
    }
  }
}

この例では、fixed_interval が 120m に設定されています。 これは、時間精度が 60 分のダウンサンプリングインデックスがデータクエリに使用されることを示しています。 次の結果が返されます。

{
  "took" : 15,
  "timed_out" : false,
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "1" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "disk_type2",
          "doc_count" : 9,
          "2" : {
            "buckets" : [
              {
                "key_as_string" : "2022-06-20T06:00:00.000Z",
                "key" : 1655704800000,
                "doc_count" : 9
              }
            ]
          }
        }
      ]
    }
  }
}