快速开始
本指南帮助初学者学习如何:
在测试环境中安装并运行Elasticsearch
向Elasticsearch添加数据
搜索和排序数据
在搜索期间从非结构化内容中提取字段
步骤1 执行 Elasticsearch 命令
建立Elasticsearch最简单的方法是在弹性云上使用Elasticsearch Service创建托管部署。如果您喜欢管理自己的测试环境，可以使用Docker安装并运行Elasticsearch
Install and run Elasticsearch
1. Install and start [Docker Desktop](https://www.docker.com/products/docker-desktop).
2. Run:
docker network create elastic
docker pull docker.elastic.co/elasticsearch/elasticsearch:7.13.2
docker run --name es01-test --net elastic -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node"
docker.elastic.co/elasticsearch/elasticsearch:7.13.2
Install and run Kibana
To analyze, visualize, and manage Elasticsearch data using an intuitive UI, install Kibana.
1. In a new terminal session, run:
docker pull docker.elastic.co/kibana/kibana:7.13.2
docker run --name kib01-test --net elastic -p 5601:5601 -e "ELASTICSEARCH_HOSTS=http://es01-test:9200" docker.elastic.co/kibana/kibana:7.13.2
2. To access Kibana, go to [http://localhost:5601](http://localhost:5601/)
步骤2 向Elasticsearch发送请求
您使用REST api向Elasticsearch发送数据和其他请求。这允许您使用任何发送HTTP请求的客户端(比如curl)与Elasticsearch进行交互。您还可以使用Kibana的控制台向Elasticsearch发送请求
Use curl
To submit an example API request, run the following curl command in a new terminal session.
curl -X GET http://localhost:9200/
Use Kibana
- Open Kibana’s main menu and go to Dev Tools > Console.
- Run the following example API request in the console:
GET /
步骤3 添加数据
您将数据作为称为文档的JSON对象添加到Elasticsearch中。Elasticsearch将这些文档存储在可搜索的索引中。
对于时间序列数据(如日志和度量)，通常将文档添加到由多个自动生成的支持索引组成的数据流中。
数据流需要一个与其名称匹配的索引模板。Elasticsearch使用这个模板来配置流的后台索引。发送到数据流的文档必须有@timestamp字段。
添加一个单独的文档
提交以下索引请求，将单个日志条目添加到logs-my_app-default数据流中。因为logs-my_app-default不存在，请求自动使用内置的logs--索引模板创建它
POST logs-my_app-default/_doc
{
"@timestamp": "2099-05-06T16:21:15.000Z",
"event": {
"original": "192.0.2.42 - - [06/May/2099:16:21:15 +0000] \"GET /images/bg.jpg HTTP/1.0\" 200 24736"
}
}
响应包含了Elasticsearch为文档生成的元数据:
- 包含文档的backing _index。Elasticsearch自动生成支持索引的名称。
- 索引内文档的唯一_id。
{
"_index": ".ds-logs-my_app-default-2099-05-06-000001",
"_type": "_doc",
"_id": "gl5MJXMBMk1dGnErnBW8",
"_version": 1,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"_seq_no": 0,
"_primary_term": 1
}
添加多个文件
使用_bulk端点在一个请求中添加多个文档。批量数据必须是新行分隔的JSON (NDJSON)。每一行必须以换行符(\n)结尾，包括最后一行。
PUT logs-my_app-default/_bulk
{ "create": { } }
{ "@timestamp": "2099-05-07T16:24:32.000Z", "event": { "original": "192.0.2.242 - - [07/May/2020:16:24:32 -0500] \"GET /images/hm_nbg.jpg HTTP/1.0\" 304 0" } }
{ "create": { } }
{ "@timestamp": "2099-05-08T16:25:42.000Z", "event": { "original": "192.0.2.255 - - [08/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638" } }
步骤4 搜索数据
建立了索引的文档可用于近乎实时的搜索。下面的搜索将匹配logs-my_app-default中的所有日志条目，并按@timestamp按降序对它们进行排序。
GET logs-my_app-default/_search
{
"query": {
"match_all": { }
},
"sort": [
{
"@timestamp": "desc"
}
]
}
默认情况下，响应的命中部分最多包括匹配搜索的前10个文档。每次命中的_source都包含在索引期间提交的原始JSON对象。
{
"took": 2,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 3,
"relation": "eq"
},
"max_score": null,
"hits": [
{
"_index": ".ds-logs-my_app-default-2099-05-06-000001",
"_type": "_doc",
"_id": "PdjWongB9KPnaVm2IyaL",
"_score": null,
"_source": {
"@timestamp": "2099-05-08T16:25:42.000Z",
"event": {
"original": "192.0.2.255 - - [08/May/2099:16:25:42 +0000] \"GET /favicon.ico HTTP/1.0\" 200 3638"
}
},
"sort": [
4081940742000
]
},
...
]
}
}
获得特定字段
对于大型文档，解析整个_source非常笨拙。要从响应中排除它，请将_source参数设置为false。相反，使用fields参数来检索所需的字段。
GET logs-my_app-default/_search
{
"query": {
"match_all": { }
},
"fields": [
"@timestamp"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
响应以平面数组的形式包含每个命中的字段值。
{
...
"hits": {
...
"hits": [
{
"_index": ".ds-logs-my_app-default-2099-05-06-000001",
"_type": "_doc",
"_id": "PdjWongB9KPnaVm2IyaL",
"_score": null,
"fields": {
"@timestamp": [
"2099-05-08T16:25:42.000Z"
]
},
"sort": [
4081940742000
]
},
...
]
}
}
搜索日期范围
如果要跨时间段或IP范围进行搜索，请使用范围查询。
GET logs-my_app-default/_search
{
"query": {
"range": {
"@timestamp": {
"gte": "2099-05-05",
"lt": "2099-05-08"
}
}
},
"fields": [
"@timestamp"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
您可以使用日期数学来定义相对时间范围。下面的查询搜索过去一天的数据，它不会匹配logs-my_app-default中的任何日志条目
GET logs-my_app-default/_search
{
"query": {
"range": {
"@timestamp": {
"gte": "now-1d/d",
"lt": "now/d"
}
}
},
"fields": [
"@timestamp"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
从非结构化内容中提取字段
You can extract runtime fields from unstructured content, such as log messages, during a search.
Use the following search to extract the source.ip runtime field from event.original. To include it in the response, add source.ip to the fields parameter.
GET logs-my_app-default/_search
{
"runtime_mappings": {
"source.ip": {
"type": "ip",
"script": """
String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ "event.original" ].value)?.sourceip;
if (sourceip != null) emit(sourceip);
"""
}
},
"query": {
"range": {
"@timestamp": {
"gte": "2099-05-05",
"lt": "2099-05-08"
}
}
},
"fields": [
"@timestamp",
"source.ip"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
组合查询
可以使用bool查询组合多个查询。下面的搜索组合了两个范围查询:一个在@timestamp上，一个在源上。ip运行时。
GET logs-my_app-default/_search
{
"runtime_mappings": {
"source.ip": {
"type": "ip",
"script": """
String sourceip=grok('%{IPORHOST:sourceip} .*').extract(doc[ "event.original" ].value)?.sourceip;
if (sourceip != null) emit(sourceip);
"""
}
},
"query": {
"bool": {
"filter": [
{
"range": {
"@timestamp": {
"gte": "2099-05-05",
"lt": "2099-05-08"
}
}
},
{
"range": {
"source.ip": {
"gte": "192.0.2.0",
"lte": "192.0.2.240"
}
}
}
]
}
},
"fields": [
"@timestamp",
"source.ip"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
聚合数据
使用聚合将数据总结为指标、统计数据或其他分析。
下面的搜索使用聚合来计算使用http.response.body.bytes运行时字段的average_response_size。聚合只在与查询匹配的文档上运行。
GET logs-my_app-default/_search
{
"runtime_mappings": {
"http.response.body.bytes": {
"type": "long",
"script": """
String bytes=grok('%{COMMONAPACHELOG}').extract(doc[ "event.original" ].value)?.bytes;
if (bytes != null) emit(Integer.parseInt(bytes));
"""
}
},
"aggs": {
"average_response_size":{
"avg": {
"field": "http.response.body.bytes"
}
}
},
"query": {
"bool": {
"filter": [
{
"range": {
"@timestamp": {
"gte": "2099-05-05",
"lt": "2099-05-08"
}
}
}
]
}
},
"fields": [
"@timestamp",
"http.response.body.bytes"
],
"_source": false,
"sort": [
{
"@timestamp": "desc"
}
]
}
响应的聚合对象包含聚合结果
{
...
"aggregations" : {
"average_response_size" : {
"value" : 12368.0
}
}
}
Explore more search options
To keep exploring, index more data to your data stream and check out Common search options.
步骤5. 清理
完成后，删除测试数据流及其备份索引
DELETE _data_stream/logs-my_app-default
- To stop your Elasticsearch and Kibana Docker containers, run:
docker stop es01-test
docker stop kib01-test
- To remove the containers and their network, run:
docker network rm elastic
docker rm es01-test
docker rm kib01-test
What’s next?
- Get the most out of your time series data by setting up data tiers and ILM. See Use Elasticsearch for time series data.
- Use Fleet and Elastic Agent to collect logs and metrics directly from your data sources and send them to Elasticsearch. See the Fleet quick start guide.
- Use Kibana to explore, visualize, and manage your Elasticsearch data. See the Kibana quick start guide.
