加载中

用例：使用 Elasticsearch 管理时序数据

Elasticsearch 提供了多种功能，可帮助您存储、管理和搜索时间序列数据，例如日志和指标。数据一旦进入 Elasticsearch，您就可以使用 Kibana 和其他 Elastic Stack 功能对数据进行分析和可视化。

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

热（hot）层和内容（content）层是必需的。温（warm）层、冷（cold）层和冻结（frozen）层是可选的。

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

内容层通常不用于时间序列数据。但是，创建系统索引以及不属于数据流的其他索引时，它是必需的。

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

  1. 登录 Elastic Cloud 控制台。
  2. 从 Elastic Cloud 主页或“托管部署（Hosted deployments）”页面添加或选择您的部署。
  3. 在部署菜单中，选择 编辑部署（Edit deployment）。
  4. 要启用数据层，请单击 添加容量（Add capacity）。

启用自动缩放

自动扩缩容（Autoscaling）会自动调整您的部署容量以满足您的存储需求。要启用自动扩缩容，请在 编辑部署（Edit deployment） 页面上选择 自动扩缩容此部署（Autoscale this deployment）。自动扩缩容仅适用于 Elastic Cloud 托管服务。

要将节点分配给某个数据层，请将相应的 节点角色 添加到该节点的 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 ]

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

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

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

found-snapshots 存储库是特定于您的集群的。要使用另一个集群的默认存储库，请参阅 Cloud 快照和恢复（Snapshot and restore）文档。

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

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

您也可以使用这些存储库类型的替代实现，例如 MinIO，只要它们完全兼容即可。使用 存储库分析（Repository analysis） API 来分析您的存储库是否适合与可搜索快照结合使用。

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

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

  • 为您配置的每个数据层都包含一个阶段。
  • 计算从滚动更新（rollover）开始进行阶段转换的阈值（即 min_age）。
  • 如果需要，在冷阶段和冻结阶段使用可搜索快照。
  • 如果需要，包含删除阶段。

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

  • 日志
  • metrics
  • synthetics（合成监控）

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

在 Kibana 中编辑策略

  1. 使用导航菜单或全局搜索字段转到 Index Lifecycle Policies 管理页面。
  2. 单击您想要编辑的策略。

您也可以使用 更新生命周期策略（update lifecycle policy） API。

				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 中创建策略

  1. 使用导航菜单或全局搜索字段转到 Index Lifecycle Policies 管理页面。
  2. 点击 Create policy。

您也可以使用 更新生命周期策略（update lifecycle policy） API。

				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 Common Schema，ECS）。ECS 字段默认与多个 Elastic Stack 功能集成。

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

在 Kibana 中创建组件模板

  1. 使用导航菜单或全局搜索字段转到 Index Management 页面。
  2. 在 索引模板（Index Templates） 选项卡中，单击 创建组件模板（Create component template）。

您也可以使用 创建组件模板（create component template） API。

				
					# 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"
  }
}

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

在 Kibana 中创建索引模板

  1. 使用导航菜单或全局搜索字段转到 Index Management 页面。
  2. 在“索引模板”（Index Templates）选项卡中，点击“创建模板”（Create template）。

您也可以使用 创建索引模板（create index template） API。包含 data_stream 对象以启用数据流。

				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 字段。

要自动创建数据流，请提交一个针对该数据流名称的索引请求。此名称必须与您的索引模板中的索引模式之一匹配。

				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 中探索和搜索您的数据，请打开主菜单并选择 发现（Discover）。请参阅 Kibana 的 发现（Discover）文档。

使用 Kibana 的 仪表板（Dashboard） 功能，以图表、表格、地图等形式可视化您的数据。请参阅 Kibana 的 仪表板（Dashboard）文档。

您也可以使用 搜索（search） API 来搜索和聚合您的数据。使用 运行时字段（runtime fields） 和 grok 模式，在搜索时从日志消息和其他非结构化内容中动态提取数据。

				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 搜索默认是同步的。对冻结数据、长跨度时间范围或大数据集的搜索可能需要更长时间。使用 异步搜索（async search） API 在后台运行搜索。有关更多搜索选项，请参阅 搜索 API（The search API）。

				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"
    }
  ]
}
© . This website operates independently and is not affiliated with or endorsed by Elasticsearch B.V. All brand names, logos, and trademarks are the property of their respective owners.