› ›

使用 Elasticsearch 存储时间序列数据

编辑

使用 Elasticsearch 存储时间序列数据

编辑

Elasticsearch 提供了帮助您存储、管理和搜索时间序列数据（例如日志和指标）的功能。将数据存储在 Elasticsearch 中后，您可以使用 Kibana 和其他 Elastic Stack 功能来分析和可视化数据。

设置数据层

编辑

Elasticsearch 的 ILM 功能使用数据层自动将较旧的数据移动到具有较低成本硬件的节点上。这有助于提高性能并降低存储成本。

热层和内容层是必需的。温层、冷层和冻结层是可选的。

在热层和温层中使用高性能节点，以便更快地索引和搜索您最新的数据。在冷层和冻结层中使用速度较慢、成本较低的节点以降低成本。

内容层通常不用于时间序列数据。但是，它需要创建系统索引和其他不是数据流一部分的索引。

设置数据层的步骤因您的部署类型而异

登录 Elasticsearch 服务控制台。
从 Elasticsearch 服务主页或部署页面添加或选择您的部署。
在您的部署菜单中，选择 编辑部署。
要启用数据层，请单击 添加容量。

启用自动伸缩

自动伸缩会自动调整部署的容量以满足您的存储需求。要启用自动伸缩，请在 编辑部署 页面上选择 自动伸缩此部署。自动伸缩仅适用于 Elasticsearch 服务。

要将节点分配给数据层，请将相应的节点角色添加到节点的 elasticsearch.yml 文件中。更改现有节点的角色需要滚动重启。

# Content tier
node.roles: [ data_content ]

# Hot tier
node.roles: [ data_hot ]

# Warm tier
node.roles: [ data_warm ]

# Cold tier
node.roles: [ data_cold ]

# Frozen tier
node.roles: [ data_frozen ]

我们建议您在冻结层中使用专用节点。如有必要，您可以将其他节点分配给多个层。

node.roles: [ data_content, data_hot, data_warm ]

为您的集群分配任何其他所需的节点角色。例如，小型集群可能具有具有多个角色的节点。

node.roles: [ master, ingest, ml, data_hot, transform ]

注册快照存储库

编辑

冷层和冻结层可以使用可搜索快照来降低本地存储成本。

要使用可搜索快照，您必须注册受支持的快照存储库。注册此存储库的步骤因您的部署类型和存储提供商而异

创建集群时，Elasticsearch 服务会自动注册默认的 found-snapshots 存储库。此存储库支持可搜索快照。

found-snapshots 存储库特定于您的集群。要使用其他集群的默认存储库，请参阅 Cloud 快照和恢复文档。

您还可以将以下任何自定义存储库类型与可搜索快照一起使用

创建或编辑索引生命周期策略

编辑

数据流将您的数据存储在多个后备索引中。ILM 使用索引生命周期策略自动将这些索引移动到您的数据层中。

如果您使用 Fleet 或 Elastic Agent，请编辑 Elasticsearch 的内置生命周期策略之一。如果您使用自定义应用程序，请创建您自己的策略。无论哪种情况，请确保您的策略

包含您已配置的每个数据层的阶段。
计算从滚动转换阶段的阈值或 min_age。
如果需要，在冷阶段和冻结阶段使用可搜索快照。
如果需要，包含删除阶段。

Fleet 和 Elastic Agent 使用以下内置生命周期策略

日志
指标
合成数据

您可以根据您的性能、弹性和保留要求自定义这些策略。

要在 Kibana 中编辑策略，请打开主菜单并转到 堆栈管理 > 索引生命周期策略。单击您要编辑的策略。

您还可以使用更新生命周期策略 API。

resp = client.ilm.put_lifecycle(
    name="logs",
    policy={
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_primary_shard_size": "50gb"
                    }
                }
            },
            "warm": {
                "min_age": "30d",
                "actions": {
                    "shrink": {
                        "number_of_shards": 1
                    },
                    "forcemerge": {
                        "max_num_segments": 1
                    }
                }
            },
            "cold": {
                "min_age": "60d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "frozen": {
                "min_age": "90d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "delete": {
                "min_age": "735d",
                "actions": {
                    "delete": {}
                }
            }
        }
    },
)
print(resp)

const response = await client.ilm.putLifecycle({
  name: "logs",
  policy: {
    phases: {
      hot: {
        actions: {
          rollover: {
            max_primary_shard_size: "50gb",
          },
        },
      },
      warm: {
        min_age: "30d",
        actions: {
          shrink: {
            number_of_shards: 1,
          },
          forcemerge: {
            max_num_segments: 1,
          },
        },
      },
      cold: {
        min_age: "60d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      frozen: {
        min_age: "90d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      delete: {
        min_age: "735d",
        actions: {
          delete: {},
        },
      },
    },
  },
});
console.log(response);

PUT _ilm/policy/logs
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_primary_shard_size": "50gb"
          }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "60d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "frozen": {
        "min_age": "90d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "delete": {
        "min_age": "735d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

要在 Kibana 中创建策略，请打开主菜单并转到 堆栈管理 > 索引生命周期策略。单击 创建策略。

您还可以使用更新生命周期策略 API。

resp = client.ilm.put_lifecycle(
    name="my-lifecycle-policy",
    policy={
        "phases": {
            "hot": {
                "actions": {
                    "rollover": {
                        "max_primary_shard_size": "50gb"
                    }
                }
            },
            "warm": {
                "min_age": "30d",
                "actions": {
                    "shrink": {
                        "number_of_shards": 1
                    },
                    "forcemerge": {
                        "max_num_segments": 1
                    }
                }
            },
            "cold": {
                "min_age": "60d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "frozen": {
                "min_age": "90d",
                "actions": {
                    "searchable_snapshot": {
                        "snapshot_repository": "found-snapshots"
                    }
                }
            },
            "delete": {
                "min_age": "735d",
                "actions": {
                    "delete": {}
                }
            }
        }
    },
)
print(resp)

const response = await client.ilm.putLifecycle({
  name: "my-lifecycle-policy",
  policy: {
    phases: {
      hot: {
        actions: {
          rollover: {
            max_primary_shard_size: "50gb",
          },
        },
      },
      warm: {
        min_age: "30d",
        actions: {
          shrink: {
            number_of_shards: 1,
          },
          forcemerge: {
            max_num_segments: 1,
          },
        },
      },
      cold: {
        min_age: "60d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      frozen: {
        min_age: "90d",
        actions: {
          searchable_snapshot: {
            snapshot_repository: "found-snapshots",
          },
        },
      },
      delete: {
        min_age: "735d",
        actions: {
          delete: {},
        },
      },
    },
  },
});
console.log(response);

PUT _ilm/policy/my-lifecycle-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_primary_shard_size": "50gb"
          }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "60d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "frozen": {
        "min_age": "90d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "found-snapshots"
          }
        }
      },
      "delete": {
        "min_age": "735d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

创建组件模板

编辑

如果您使用 Fleet 或 Elastic Agent，请跳至搜索和可视化您的数据。Fleet 和 Elastic Agent 使用内置模板为您创建数据流。

如果您使用自定义应用程序，则需要设置您自己的数据流。数据流需要匹配的索引模板。在大多数情况下，您可以使用一个或多个组件模板来组合此索引模板。您通常为映射和索引设置使用单独的组件模板。这使您可以在多个索引模板中重用组件模板。

创建组件模板时，请包含

用于 @timestamp 字段的 date 或 date_nanos 映射。如果您没有指定映射，Elasticsearch 会将 @timestamp 映射为具有默认选项的 date 字段。
您在 index.lifecycle.name 索引设置中的生命周期策略。

映射字段时使用 Elastic 通用模式 (ECS)。ECS 字段默认情况下与多个 Elastic Stack 功能集成。

如果您不确定如何映射字段，请使用运行时字段在搜索时从非结构化内容中提取字段。例如，您可以将日志消息索引到 wildcard 字段，然后在搜索期间从该字段中提取 IP 地址和其他数据。

要在 Kibana 中创建组件模板，请打开主菜单并转到 堆栈管理 > 索引管理。在 索引模板 视图中，单击 创建组件模板。

您还可以使用创建组件模板 API。

resp = client.cluster.put_component_template(
    name="my-mappings",
    template={
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date",
                    "format": "date_optional_time||epoch_millis"
                },
                "message": {
                    "type": "wildcard"
                }
            }
        }
    },
    meta={
        "description": "Mappings for @timestamp and message fields",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp)

resp1 = client.cluster.put_component_template(
    name="my-settings",
    template={
        "settings": {
            "index.lifecycle.name": "my-lifecycle-policy"
        }
    },
    meta={
        "description": "Settings for ILM",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp1)

response = client.cluster.put_component_template(
  name: 'my-mappings',
  body: {
    template: {
      mappings: {
        properties: {
          "@timestamp": {
            type: 'date',
            format: 'date_optional_time||epoch_millis'
          },
          message: {
            type: 'wildcard'
          }
        }
      }
    },
    _meta: {
      description: 'Mappings for @timestamp and message fields',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response

response = client.cluster.put_component_template(
  name: 'my-settings',
  body: {
    template: {
      settings: {
        'index.lifecycle.name' => 'my-lifecycle-policy'
      }
    },
    _meta: {
      description: 'Settings for ILM',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response

const response = await client.cluster.putComponentTemplate({
  name: "my-mappings",
  template: {
    mappings: {
      properties: {
        "@timestamp": {
          type: "date",
          format: "date_optional_time||epoch_millis",
        },
        message: {
          type: "wildcard",
        },
      },
    },
  },
  _meta: {
    description: "Mappings for @timestamp and message fields",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);

const response1 = await client.cluster.putComponentTemplate({
  name: "my-settings",
  template: {
    settings: {
      "index.lifecycle.name": "my-lifecycle-policy",
    },
  },
  _meta: {
    description: "Settings for ILM",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response1);

# Creates a component template for mappings
PUT _component_template/my-mappings
{
  "template": {
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date",
          "format": "date_optional_time||epoch_millis"
        },
        "message": {
          "type": "wildcard"
        }
      }
    }
  },
  "_meta": {
    "description": "Mappings for @timestamp and message fields",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

# Creates a component template for index settings
PUT _component_template/my-settings
{
  "template": {
    "settings": {
      "index.lifecycle.name": "my-lifecycle-policy"
    }
  },
  "_meta": {
    "description": "Settings for ILM",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

创建索引模板

编辑

使用您的组件模板创建索引模板。指定

一个或多个与数据流名称匹配的索引模式。我们建议使用我们的数据流命名方案。
该模板已启用数据流。
包含您的映射和索引设置的任何组件模板。
高于 200 的优先级，以避免与内置模板发生冲突。请参阅避免索引模式冲突。

要在 Kibana 中创建索引模板，请打开主菜单并转到 堆栈管理 > 索引管理。在 索引模板 视图中，单击 创建模板。

您还可以使用创建索引模板 API。包含 data_stream 对象以启用数据流。

resp = client.indices.put_index_template(
    name="my-index-template",
    index_patterns=[
        "my-data-stream*"
    ],
    data_stream={},
    composed_of=[
        "my-mappings",
        "my-settings"
    ],
    priority=500,
    meta={
        "description": "Template for my time series data",
        "my-custom-meta-field": "More arbitrary metadata"
    },
)
print(resp)

response = client.indices.put_index_template(
  name: 'my-index-template',
  body: {
    index_patterns: [
      'my-data-stream*'
    ],
    data_stream: {},
    composed_of: [
      'my-mappings',
      'my-settings'
    ],
    priority: 500,
    _meta: {
      description: 'Template for my time series data',
      "my-custom-meta-field": 'More arbitrary metadata'
    }
  }
)
puts response

const response = await client.indices.putIndexTemplate({
  name: "my-index-template",
  index_patterns: ["my-data-stream*"],
  data_stream: {},
  composed_of: ["my-mappings", "my-settings"],
  priority: 500,
  _meta: {
    description: "Template for my time series data",
    "my-custom-meta-field": "More arbitrary metadata",
  },
});
console.log(response);

PUT _index_template/my-index-template
{
  "index_patterns": ["my-data-stream*"],
  "data_stream": { },
  "composed_of": [ "my-mappings", "my-settings" ],
  "priority": 500,
  "_meta": {
    "description": "Template for my time series data",
    "my-custom-meta-field": "More arbitrary metadata"
  }
}

将数据添加到数据流

编辑

索引请求将文档添加到数据流。这些请求必须使用 op_type 为 create。文档必须包含 @timestamp 字段。

要自动创建数据流，请提交一个将目标指向流名称的索引请求。此名称必须与您的索引模板的索引模式之一匹配。

resp = client.bulk(
    index="my-data-stream",
    operations=[
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:21:15.000Z",
            "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
        },
        {
            "create": {}
        },
        {
            "@timestamp": "2099-05-06T16:25:42.000Z",
            "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638"
        }
    ],
)
print(resp)

resp1 = client.index(
    index="my-data-stream",
    document={
        "@timestamp": "2099-05-06T16:21:15.000Z",
        "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
    },
)
print(resp1)

response = client.bulk(
  index: 'my-data-stream',
  body: [
    {
      create: {}
    },
    {
      "@timestamp": '2099-05-06T16:21:15.000Z',
      message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
    },
    {
      create: {}
    },
    {
      "@timestamp": '2099-05-06T16:25:42.000Z',
      message: '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638'
    }
  ]
)
puts response

response = client.index(
  index: 'my-data-stream',
  body: {
    "@timestamp": '2099-05-06T16:21:15.000Z',
    message: '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736'
  }
)
puts response

const response = await client.bulk({
  index: "my-data-stream",
  operations: [
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:21:15.000Z",
      message:
        '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
    },
    {
      create: {},
    },
    {
      "@timestamp": "2099-05-06T16:25:42.000Z",
      message:
        '192.0.2.255 - - [06/May/2099:16:25:42 +0000] "GET /favicon.ico HTTP/1.0" 200 3638',
    },
  ],
});
console.log(response);

const response1 = await client.index({
  index: "my-data-stream",
  document: {
    "@timestamp": "2099-05-06T16:21:15.000Z",
    message:
      '192.0.2.42 - - [06/May/2099:16:21:15 +0000] "GET /images/bg.jpg HTTP/1.0" 200 24736',
  },
});
console.log(response1);

PUT my-data-stream/_bulk
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:21:15.000Z", "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736" }
{ "create":{ } }
{ "@timestamp": "2099-05-06T16:25:42.000Z", "message": "192.0.2.255 - - [06/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638" }

POST my-data-stream/_doc
{
  "@timestamp": "2099-05-06T16:21:15.000Z",
  "message": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
}

搜索和可视化您的数据

编辑

要在 Kibana 中浏览和搜索您的数据，请打开主菜单并选择发现。请参阅 Kibana 的发现文档。

使用 Kibana 的 仪表盘 功能以图表、表格、地图等方式可视化您的数据。请参阅 Kibana 的仪表盘文档。

您还可以使用搜索 API 搜索和聚合数据。使用运行时字段和 grok 模式在搜索时动态地从日志消息和其他非结构化内容中提取数据。

resp = client.search(
    index="my-data-stream",
    runtime_mappings={
        "source.ip": {
            "type": "ip",
            "script": "\n        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ \"message\" ].value)?.sourceip;\n        if (sourceip != null) emit(sourceip);\n      "
        }
    },
    query={
        "bool": {
            "filter": [
                {
                    "range": {
                        "@timestamp": {
                            "gte": "now-1d/d",
                            "lt": "now/d"
                        }
                    }
                },
                {
                    "range": {
                        "source.ip": {
                            "gte": "192.0.2.0",
                            "lte": "192.0.2.255"
                        }
                    }
                }
            ]
        }
    },
    fields=[
        "*"
    ],
    source=False,
    sort=[
        {
            "@timestamp": "desc"
        },
        {
            "source.ip": "desc"
        }
    ],
)
print(resp)

const response = await client.search({
  index: "my-data-stream",
  runtime_mappings: {
    "source.ip": {
      type: "ip",
      script:
        "\n        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ \"message\" ].value)?.sourceip;\n        if (sourceip != null) emit(sourceip);\n      ",
    },
  },
  query: {
    bool: {
      filter: [
        {
          range: {
            "@timestamp": {
              gte: "now-1d/d",
              lt: "now/d",
            },
          },
        },
        {
          range: {
            "source.ip": {
              gte: "192.0.2.0",
              lte: "192.0.2.255",
            },
          },
        },
      ],
    },
  },
  fields: ["*"],
  _source: false,
  sort: [
    {
      "@timestamp": "desc",
    },
    {
      "source.ip": "desc",
    },
  ],
});
console.log(response);

GET my-data-stream/_search
{
  "runtime_mappings": {
    "source.ip": {
      "type": "ip",
      "script": """
        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ "message" ].value)?.sourceip;
        if (sourceip != null) emit(sourceip);
      """
    }
  },
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gte": "now-1d/d",
              "lt": "now/d"
            }
          }
        },
        {
          "range": {
            "source.ip": {
              "gte": "192.0.2.0",
              "lte": "192.0.2.255"
            }
          }
        }
      ]
    }
  },
  "fields": [
    "*"
  ],
  "_source": false,
  "sort": [
    {
      "@timestamp": "desc"
    },
    {
      "source.ip": "desc"
    }
  ]
}

Elasticsearch 搜索默认情况下是同步的。跨冻结数据、长时间范围或大型数据集的搜索可能需要更长时间。使用异步搜索 API可在后台运行搜索。有关更多搜索选项，请参阅搜索 API。

resp = client.async_search.submit(
    index="my-data-stream",
    runtime_mappings={
        "source.ip": {
            "type": "ip",
            "script": "\n        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ \"message\" ].value)?.sourceip;\n        if (sourceip != null) emit(sourceip);\n      "
        }
    },
    query={
        "bool": {
            "filter": [
                {
                    "range": {
                        "@timestamp": {
                            "gte": "now-2y/d",
                            "lt": "now/d"
                        }
                    }
                },
                {
                    "range": {
                        "source.ip": {
                            "gte": "192.0.2.0",
                            "lte": "192.0.2.255"
                        }
                    }
                }
            ]
        }
    },
    fields=[
        "*"
    ],
    source=False,
    sort=[
        {
            "@timestamp": "desc"
        },
        {
            "source.ip": "desc"
        }
    ],
)
print(resp)

const response = await client.asyncSearch.submit({
  index: "my-data-stream",
  runtime_mappings: {
    "source.ip": {
      type: "ip",
      script:
        "\n        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ \"message\" ].value)?.sourceip;\n        if (sourceip != null) emit(sourceip);\n      ",
    },
  },
  query: {
    bool: {
      filter: [
        {
          range: {
            "@timestamp": {
              gte: "now-2y/d",
              lt: "now/d",
            },
          },
        },
        {
          range: {
            "source.ip": {
              gte: "192.0.2.0",
              lte: "192.0.2.255",
            },
          },
        },
      ],
    },
  },
  fields: ["*"],
  _source: false,
  sort: [
    {
      "@timestamp": "desc",
    },
    {
      "source.ip": "desc",
    },
  ],
});
console.log(response);

POST my-data-stream/_async_search
{
  "runtime_mappings": {
    "source.ip": {
      "type": "ip",
      "script": """
        String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ "message" ].value)?.sourceip;
        if (sourceip != null) emit(sourceip);
      """
    }
  },
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gte": "now-2y/d",
              "lt": "now/d"
            }
          }
        },
        {
          "range": {
            "source.ip": {
              "gte": "192.0.2.0",
              "lte": "192.0.2.255"
            }
          }
        }
      ]
    }
  },
  "fields": [
    "*"
  ],
  "_source": false,
  "sort": [
    {
      "@timestamp": "desc"
    },
    {
      "source.ip": "desc"
    }
  ]
}

«调整分片大小自动伸缩 »