EFK架构部署

说明：全篇elastic是基于Ubuntu系统安装的，与Centos差别不是太大，下载软件连接可去官网下载

ElasitcStack概述

1.什么是ElasticStack？

所谓的ElasticStack别名为elk stack。

ELK指的是三个技术栈:

- ElasticSearch，简称： es
数据库，应用场景为数据的快速检索。但凡和搜索框相关的，都会用ES进行数据的查询。

- Logstash:	
	采集数据，日志聚合，处理数据，将数据写入到ES存储库。
	
- Kibana:
	用于出图展示数据。
	
	
由于Logstash比较重量级，采集日志时启动较慢，使用JRuby语言研发，启动时需要先启动虚拟机。因此官方开发了beats组件，beats组件是一系列beat的集合，比如filebeat就是一个常用的beat组件，多用于文本日志采集。

2.常用的架构

EFK:
- ElasticSearch
- Filebeat:
采集数据写入ES集群。
- Kibana

ELFK:
- ElasticSearch
- Filebeat:
- Logstash
- Kibana

ELKF+KAFKA:
- zookeeper
- kafka

ElasticSearch

1.什么是ElasticSearch

Elasticsearch 是一个分布式、RESTful 风格的搜索和数据分析引擎，能够解决不断涌现出的各种用例。作为 Elastic Stack 的核心，Elasticsearch 会集中存储您的数据，让您飞快完成搜索，微调相关性，进行强大的分析，并轻松缩放规模。ES的核心作用就是存储数据和快速进行模糊查询。

2.安装部署ES

单点部署ES 一般用于测试

`本机ip：10.0.0.211`
1.下载ES
[root@elk01 ~]# wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.23-amd64.deb

2.安装ES
[root@elk01 ~]# dpkg -i elasticsearch-7.17.23-amd64.deb

3.修改配置文件
[root@elk01 ~]# vim /etc/elasticsearch/elasticsearch.yml 
cluster.name: elk
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
network.host: 0.0.0.0
http.port: 9200
transport.port: 9300
discovery.type: single-node


`相关参数说明:`
	cluster.name: 				#指定ES集群的名称
	path.data: 						#指定数据目录
	path.logs: 						#指定日志目录
	network.host: 				#指定宿主机的监听IP。
	http.port: 9200				#对外提供http|https接口的端口。
	transport.port: 9300	#集群内部数据传输的端口，会优先启动该端口选举master后在启动9200端口。
	discovery.type: 			#指定ES集群的类型，single-node表示的是单点。
	
4.启动ES并设置开机自启
[root@elk01 ~]# systemctl enable --now elasticsearch

5.检查监听的端口号
[root@elk01 ~]# ss -lntup |egrep '9[23]00'
tcp6       0      0 :::9200        :::*       LISTEN      6431/java
tcp6       0      0 :::9300        :::*       LISTEN      6431/java

6.测试访问
[root@elk01:0 ~]# curl 10.0.0.211:9200
{
  "name" : "elk01",
  "cluster_name" : "elk",
  "cluster_uuid" : "vV46E9vqRK-MxDs6I-3ceg",
  "version" : {
    "number" : "7.17.23",
    "build_flavor" : "default",
    "build_type" : "deb",
    "build_hash" : "61d76462eecaf09ada684d1b5d319b5ff6865a83",
    "build_date" : "2024-07-25T14:37:42.448799567Z",
    "build_snapshot" : false,
    "lucene_version" : "8.11.3",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

#查看集群节点
[root@elk01:0 ~]# curl 10.0.0.211:9200/_cat/nodes
10.0.0.211 19 91 11 0.10 0.23 0.11 cdfhilmrstw * elk01

#显示表头信息    *代表master节点
[root@elk01:0 ~]# curl 10.0.0.211:9200/_cat/nodes?v
ip         heap.percent ram.percent cpu load_1m load_5m load_15m node.role   master name
10.0.0.211           20          91   0    0.06    0.21     0.11 cdfhilmrstw *      elk01

如何卸载ES？

1.停止ES服务
[root@elk91 ~]# systemctl disable --now elasticsearch.service 

2.删除数据
[root@elk91 ~]# \rm -rf  /var/{lib,log}/elasticsearch/ /tmp/*  /etc/elasticsearch

3.删除软件
[root@elk91 ~]# sudo yum remove elasticsearch
[root@elk91 ~]# sudo rpm -e elasticsearch

3.部署ES集群

主机名	WanIP	LanIP
elk01	10.0.0.211	172.16.1.211
elk02	10.0.0.222	172.16.1.222
elk03	10.0.0.233	172.16.1.233

1.所有节点安装ElasticSearch

1.下载ES
[root@elk01 ~]# wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.23-amd64.deb
[root@elk01 ~]# for i in 212 213;do scp elasticsearch-7.17.23-x86_64.rpm root@172.16.1.$i:/root ;done

2.安装ES
[root@elk01 ~]#dpkg -i elasticsearch-7.17.23-amd64.deb

2.修改配置文件

[root@elk01 ~]# vim /etc/elasticsearch/elasticsearch.yml 
cluster.name: elk
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
network.host: 0.0.0.0
http.port: 9200
transport.port: 9300
discovery.seed_hosts: ["10.0.0.211", "10.0.0.212","10.0.0.213"]
cluster.initial_master_nodes: ["10.0.0.211", "10.0.0.212","10.0.0.213"]


`相关参数说明:`
	discovery.seed_hosts:					#当前ES集群的主机列表。
	cluster.initial_master_nodes:	#集群启动时首次参与master选举的节点列表。

3.将配置文件拷贝到其他节点

[root@elk01 ~]# scp /etc/elasticsearch/elasticsearch.yml root@172.16.1.212:/etc/elasticsearch/
[root@elk01 ~]# scp /etc/elasticsearch/elasticsearch.yml root@172.16.1.213:/etc/elasticsearch/

**4.启动ES集群 **

[root@elk01 ~]# systemctl enable --now elasticsearch.service 
[root@elk02 ~]# systemctl enable --now elasticsearch.service 
[root@elk03 ~]# systemctl enable --now elasticsearch.service 

5.查看端口是否监听集群

[root@elk01 ~]# ss -lntup |egrep '9[23]00'
tcp6       0      0 :::9200        :::*       LISTEN      6431/java
tcp6       0      0 :::9300        :::*       LISTEN      6431/java

6.查看集群节点信息

[root@elk01:0 ~]# curl 10.0.0.213:9200/_cat/nodes
10.0.0.212  9 94  8 0.46 0.49 0.31 cdfhilmrstw - elk02
10.0.0.211 25 95 21 0.56 0.47 0.25 cdfhilmrstw * elk01
10.0.0.213 28 90 25 1.21 0.78 0.43 cdfhilmrstw - elk03

#elk02是主节点
#查看name和uuid
[root@elk01 ~]# curl -s 10.0.0.211:9200|egrep '_name|_uuid'
  "cluster_name" : "my-elk",
  "cluster_uuid" : "pI91AngOTf2G4LYIUzd9FA"
  
  
#查看所有节点？
[root@elk01 ~]# for i in `seq 211 213`;do curl -s 10.0.0.$i:9200|egrep '_name|_uuid';done
  "cluster_name" : "my-elk",
  "cluster_uuid" : "pI91AngOTf2G4LYIUzd9FA",
  "cluster_name" : "my-elk",
  "cluster_uuid" : "pI91AngOTf2G4LYIUzd9FA",
  "cluster_name" : "my-elk",
  "cluster_uuid" : "pI91AngOTf2G4LYIUzd9FA",

7.ES集群写入和读取数据

#写入数据  （test是索引，自定义）
curl -X POST -H "Content-Type:application/json" 10.0.0.211:9200/test/_doc -d '{"name":"zhang3","age":"18"}'

#读取数据（jq需要安装，格式化输出，212/213也能读取到数据，因为是es集群）
[root@elk01 ~]# curl -s 10.0.0.211:9200/test/_search | jq .hits
{
  "total": {
    "value": 1,
    "relation": "eq"
  },
  "max_score": 1,
  "hits": [
    {
      "_index": "test",
      "_type": "_doc",
      "_id": "ppKQ0ZEB6pCPlAoKbzPE",
      "_score": 1,
      "_source": {
        "name": "zhang3",
        "age": "18"
      }
    }
  ]
}

kibana读取数据

1.kibana概述

使用 Kibana 针对大规模数据快速运行数据分析，以实现可观测性、安全和搜索。对来自任何来源的任何数据进行全面透彻的分析，从威胁情报到搜索分析，从日志到应用程序监测，不一而足。说白了，kibana就是从ES查询数据并展示数据。

2.部署kibana

1.下载 & 安装kibana

[root@elk01 ~]# wget https://artifacts.elastic.co/downloads/kibana/kibana-7.17.23-amd64.deb

[root@elk01 ~]# dpkg -i kibana-7.17.23-amd64.deb 

2.修改配置文件

[root@elk91 ~]# vim /etc/kibana/kibana.yml 
server.port: 5601
server.host: "0.0.0.0"
server.name: "kibana"
elasticsearch.hosts: ["http://10.0.0.211:9200","http://10.0.0.212:9200","http://10.0.0.213:9200"]
i18n.locale: "zh-CN"

3.启动kibana

[root@elk01 ~]# systemctl enable --now kibana

#查看端口是否启动
[root@elk01 ~]# ss  -lntup|grep 5601
tcp        0      0 0.0.0.0:5601      0.0.0.0:*     LISTEN      9696/node  


#浏览器访问
10.0.0.211:5601

索引管理就有我们之前创建的数据

如何查询那个数据呢？

4.kibana查询ES数据

1.准备测试数据

curl -X POST -H "Content-Type:application/json" 10.0.0.211:9200/_bulk -d '
{ "index" : { "_index" : "student1"} }
{"name":"李四","hobby":["学习","玩手机","看美女"]}
{ "index" : { "_index" : "student2"} }
{"name":"张三","hobby":["大长腿","学习"]}
'

#终端输出如下即OK
{"took":780,"errors":false,"items":[{"index":{"_index":"student","_type":"_doc","_id":"YWacOJIBCijr5nh3GosC","_version":1,"result":"created","_shards":{"total":2,"successful":2,"failed":0},"_seq_no":0,"_primary_term":1,"status":201}},{"index":{"_index":"student","_type":"_doc","_id":"YmacOJIBCijr5nh3GosC","_version":1,"result":"created","_shards":{"total":2,"successful":2,"failed":0},"_seq_no":1,"_primary_term":1,"status":201}}]}


#api调取方法--官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/7.17/docs-bulk.html

#浏览器再次访问  kibana  ，发现索引管理会多出来一个我们创建的student索引

2.创建索引模式

kibana的索引模式用于匹配ES集群的一个或多个索引。 student*

3.kibana查询数据

以字段查询

使用json格式查询

其它方式查询集群颜色

1.通过es-head组件查询（一个插件）

2.基于官方的API查询

[root@elk01:1 ~]# curl -s 10.0.0.211:9200/_cat/indices?v

[root@elk01:1 ~]# curl -s 10.0.0.211:9200/_cat/indices|awk '{print $1}'
green
green
green
green
green

Filebeat采集数据

filebeat
无论您是从安全设备、云、容器、主机还是 OT 进行数据收集，Filebeat 都将为您提供一种轻量型方法，用于转发和汇总日志与文件，让简单的事情不再繁杂。说白了，filebeat的核心就是采集日志并将日志发送到我们期望的目的端。本质核心就是input(数据从哪来)和output(数据到哪去)。

==filebeat默认是按行读取数据==

部署filebeat

Filebeat Reference 官方文档

1.下载filebeat包
[root@elk01 ~]# wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.17.23-amd64.deb

2.安装filebeat 
[root@elk01 ~]# dpkg -i filebeat-7.17.23-amd64.deb 

采集数据终端到终端

#写法见官方，直接复制使用  (在 inputs 和 Output 里选择相应的方式)
'https://www.elastic.co/guide/en/beats/filebeat/7.17/configuring-howto-filebeat.html'
1.编写filebeat的配置文件      （inputs>Stdin  标准输入    output>Console  终端输出）
[root@elk01 ~]# cat /etc/filebeat/01-stdin-stdout.yaml 
# 数据从标准输入来
filebeat.inputs:
- type: stdin

# 数据到终点
output.console:
  pretty: true
  
#2.启动filebeat实例
相关参数说 ：
	-c <config>:
		指定的是配置文件 
	-e :
		开启调试模式，将日志输出到终端。

[root@elk01:1 ~]# filebeat -e -c /etc/filebeat/01-stdin-stdout.yaml 
#终端我输入的
66666666666666666666666666666666

#输出内容：很明显，数据被封装到了一个"message"字段中。
{
  "@timestamp": "2024-09-28T13:23:12.146Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.17.23"
  },
  "input": {
    "type": "stdin"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "host": {
    "name": "elk01"
  },
  "agent": {
    "type": "filebeat",
    "version": "7.17.23",
    "hostname": "elk01",
    "ephemeral_id": "13686d12-e255-4c48-a609-dcc4b8331c44",
    "id": "9e9b0f4e-cbe0-4e5b-bab2-501e50f67620",
    "name": "elk01"
  },
  "log": {
    "offset": 0,
    "file": {
      "path": ""
    }
  },
  "message": "66666666666666666666666666666666"
}
2024-09-28T21:23:13.147+0800    ERROR   file/states.go:125      State for  should have been dropped, but couldn't as state is not finished.

filebeat监听tcp端口

#1。编写filebeat配置文件       （inputs>tcp   指定tcp端口）
[root@elk01 ~]# cat /etc/filebeat/02-tcp-console.yaml 
# 数据从监听的指定tcp端口来
filebeat.inputs:
- type: tcp
  host: "0.0.0.0:9000"
  
#数据到终点
output.console:
  pretty: true
  
#2.启动filebeat实例
[root@elk01:1 ~]# filebeat -e -c /etc/filebeat/02-tcp-console.yaml 

#3.发送测试数据
[root@elk03:0 ~]# echo "hello world" | nc 10.0.0.211 9000


#4.观察filebeat是否采集到数据
{
  "@timestamp": "2024-09-28T13:33:01.962Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.17.23"
  },
  "input": {
    "type": "tcp"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "host": {
    "name": "elk01"
  },
  "agent": {
    "name": "elk01",
    "type": "filebeat",
    "version": "7.17.23",
    "hostname": "elk01",
    "ephemeral_id": "a43721b4-8160-4985-9f6f-67c0178affa7",
    "id": "9e9b0f4e-cbe0-4e5b-bab2-501e50f67620"
  },
  "message": "hello world",
  "log": {
    "source": {
      "address": "10.0.0.213:47174"
    }
  }
}

采集本地文件输出数据到终端

1.编写filebeat配置文件    （inputs>log）
[root@elk01 ~]# cat /etc/filebeat/03-log-console.yaml
# 数据从监听的指定文件来
filebeat.inputs:
- type: log
  paths:
    - /tmp/test.log

# 数据到终点
output.console:
  pretty: true
  
2.启动filebeat实例
[root@elk01:1 ~]# filebeat -e -c /etc/filebeat/03-log-console.yaml

3.写入测试数据(另开一个终端写入)
[root@elk01:0 ~]# echo '999' > /tmp/test.log

4.观察终端输出
{
  "@timestamp": "2024-09-28T13:48:53.645Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.17.23"
  },
  "log": {
    "offset": 0,           #注意这个字段
    "file": {
      "path": "/tmp/test.log"
    }
  },
  "message": "hello world",
  "input": {
    "type": "log"
  },
  "agent": {
    "id": "9e9b0f4e-cbe0-4e5b-bab2-501e50f67620",
    "name": "elk01",
    "type": "filebeat",
    "version": "7.17.23",
    "hostname": "elk01",
    "ephemeral_id": "10485e22-a3e8-40f0-bbb9-c194f9541bf4"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "host": {
    "name": "elk01"
  }
}

5.验证filebeat采集数据是按行读取 (注意offset 字段变化)

#再次写入数据观察采集情况
[root@elk01:0 ~]# echo 'hello world' >> /tmp/test.log

#观察offset（偏移量）
{
  "@timestamp": "2024-09-28T13:54:03.721Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.17.23"
  },
  "log": {
    "offset": 4,    #为啥是四个偏移量？
    "file": {
      "path": "/tmp/test.log"
    }
  },
  "message": "999hello world",
  "input": {
    "type": "log"
  },
  "agent": {
    "hostname": "elk01",
    "ephemeral_id": "10485e22-a3e8-40f0-bbb9-c194f9541bf4",
    "id": "9e9b0f4e-cbe0-4e5b-bab2-501e50f67620",
    "name": "elk01",
    "type": "filebeat",
    "version": "7.17.23"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "host": {
    "name": "elk01"
  }
}

#因为上次999三个之后有个特殊字符--换行符，所以offset之前采集的偏移量是4个

如何验证filebeat是按行采集数据的？

[root@elk01:0 ~]# echo -n 'ABCD' >> /tmp/test.log      -n 不换行

观察终端是否有采集数据,很明显没有采集

6.重启filebeat实例

停止filebeat   Ctrl+c
再次执行实例    filebeat -e -c /etc/filebeat/03-log-console.yaml

#此时我们发现filebeat并不会采集数据，因为filebeat已经采集过了，
#filebeat是如何知道有没有采集过数据？往上翻看加载信息如下图

Finished loading transaction log file for '/var/lib/filebeat/registry/filebeat'. Active transaction id=63

这个表示的是，在采集数据之前先加载'/var/lib/filebeat/registry/filebeat'下的事务日志log.json，里面记录了已经采集过的信息，那么我们前去查看下，

这里我们查看下最后一条数据
[root@elk01:1 ~]# tail -1 /var/lib/filebeat/registry/filebeat/log.json 
{"k":"filebeat::logs::native::788078-64768","v":{"id":"native::788078-64768","offset":24,"timestamp":[2061673550511,1727837342],"ttl":-1,"prev_id":"","source":"/tmp/test.log","type":"log","FileStateOS":{"inode":788078,"device":64768},"identifier_name":"native"}}

#数据有个offset字段，这里记录的是上一次采集过的偏移量，那么我们就来查看下这个日志的偏移量是不是24
[root@elk01:1 ~]# cat /tmp/test.log 
999
hello world
ABCD666

#其中每个换行之后都有一个换行符占一个偏移量，所以总过是24，那么这个事务日志记录的是没有错的。当我们删除这个事务日志log.json时候，重新执行filebeat实例，便会重新采集数据

7.自定义采集

#根据步骤6中重新filebeat实例所述，我们自定义采集便可以根据设置偏移量来达到自定义采集数据
#修改事务日志文件如下，只修改偏移量，需求是只采集ABCD666及以后的数据
[root@elk01:1 ~]# cat /tmp/test.log 
999
hello world
ABCD666  #将offset设置为16，记得数换行符


[root@elk01:1 ~]# vim /var/lib/filebeat/registry/filebeat/log.json 
{"op":"set","id":65}
{"k":"filebeat::logs::native::788078-64768","v":{"id":"native::788078-64768","offset":16,"timestamp":[2061673550511,1727837342],"ttl":-1,"prev_id":"","source":"/tmp/test.log","type":"log","FileStateOS":{"inode":788078,"device":64768},"identifier_name":"native"}}

#重新执行实例
[root@elk01:0 ~]# filebeat -e -c /etc/filebeat/03-log-console.yaml 
{
  "@timestamp": "2024-10-02T03:12:45.360Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.17.23"
  },
  "agent": {
    "version": "7.17.23",
    "hostname": "elk01",
    "ephemeral_id": "c8d05474-6175-4dd1-93e8-933cb7760b92",
    "id": "9e9b0f4e-cbe0-4e5b-bab2-501e50f67620",
    "name": "elk01",
    "type": "filebeat"
  },
  "log": {
    "offset": 16,
    "file": {
      "path": "/tmp/test.log"
    }
  },
  "message": "ABCD666",
  "input": {
    "type": "log"
  },
  "ecs": {
    "version": "1.12.0"
  },
  "host": {
    "name": "elk01"
  }
}

案例

采集nginx日志案例

1.安装nginx   （可以先尝试跳过2步骤，看看不修改配置文件的输出，和修改之后的区别）
[root@elk01:1 ~]# apt -y install nginx


2.将nginx采集的数据以json格式展示，修改nginx的配置文件（格式化展示，方便分析数据，访问ip，状态码等）
[root@elk01:1 ~]# vim /etc/nginx/nginx.conf 
#http模块里添加如下，注释官方的   access_log /var/log/nginx/access.log
···
	  log_format nginx_json '{"timestamp":"$time_iso8601",'
					       '"vhost":"$server_addr",'
					       '"clientip":"$remote_addr",'
					       '"SendBytes":$body_bytes_sent,'
					       '"responsetime":$request_time,'
					       '"upstreamtime":"$upstream_response_time",'
					       '"upstreamhost":"$upstream_addr",'
					       '"http_host":"$host",'
					       '"uri":"$uri",'
					       '"domain":"$host",'
					       '"xff":"$http_x_forwarded_for",'
					       '"referer":"$http_referer",'
					       '"tcp_xff":"$proxy_protocol_addr",'
					       '"http_user_agent":"$http_user_agent",'
					       '"status":"$status"}';
	 access_log /var/log/nginx/access.log nginx_json;
#2.1 检查nginx语法是否有问题
[root@elk01:1 ~]# nginx -t 
nginx: the configuration file /etc/nginx/nginx.conf syntax is ok
nginx: configuration file /etc/nginx/nginx.conf test is successful
#2.2 重新nginx
[root@elk01:1 ~]# systemctl restart nginx


3.编写filebeat配置文件    （inputs>log）
[root@elk01:1 ~]# cat /etc/filebeat/04-nginx-es.yaml 
filebeat.inputs:
- type: log
  json:
    #当json.keys_under_root的值为true时，会自动解析json格式，并将字段放在顶级字段中
    keys_under_root: true
    #当json格式解析错误时，可以将错误信息写入到event事件(指的是filebeat采集的每一条数据)中。
    add_error_key: true
    #如果采集的字段和filebeat自己的字段冲突，则覆盖数据
    overwirte_kes: true
  paths:
    - /var/log/nginx/access.log* 

output.console:
  pretty: true
  
2.启动filebeat实例
[root@elk01:1 ~]# filebeat -e -c /etc/filebeat/03-log-console.yaml
3.另开一个终端测试
[root@elk01:1 ~]# curl 10.0.0.211
#查看filebeat收集日志
1.这是未配置  json.keys_under_root: true  的展示
  "message": "{\"timestamp\":\"2024-10-02T11:40:24+08:00\",\"vhost\":\"10.0.0.211\",\"clientip\":\"10.0.0.211\",\"SendBytes\":612,\"responsetime\":0.000,\"upstreamtime\":\"-\",\"upstreamhost\":\"-\",\"http_host\":\"10.0.0.211\",\"uri\":\"/index.nginx-debian.html\",\"domain\":\"10.0.0.211\",\"xff\":\"-\",\"referer\":\"-\",\"tcp_xff\":\"-\",\"http_user_agent\":\"curl/7.81.0\",\"status\":\"200\"}",
  
2.这是配置json.keys_under_root: true的展示  （ip，状态码，时间戳都以顶级字段展示了，更加方便读取分析数据）
  "json": {
    "clientip": "10.0.0.211",
    "domain": "10.0.0.211",
    "timestamp": "2024-10-02T11:55:24+08:00",
    "referer": "-",
    "vhost": "10.0.0.211",
    "http_host": "10.0.0.211",
    "tcp_xff": "-",
    "xff": "-",
    "uri": "/index.nginx-debian.html",
    "http_user_agent": "curl/7.81.0",
    "responsetime": 0,
    "upstreamhost": "-",
    "SendBytes": 612,
    "upstreamtime": "-",
    "status": "200"
  }
}

ES架构采集nginx日志并分析

#1.编写filebeat吸入ES集群配置文件   （选择output>Elasticsearch复制官方案例修改）
[root@elk01:1 ~]# cat /etc/filebeat/05-nginx-es.yaml 
filebeat.inputs:
- type: log
  json:
    keys_under_root: true
    add_error_key: true
    overwirte_kes: true
  paths:
    - /var/log/nginx/access.log* 

output.elasticsearch:
  hosts: ["http://10.0.0.211:9200","http://10.0.0.212:9200","http://10.0.0.213:9200"] 
  
#2.启动filebeat实例
[root@elk01:1 ~]# filebeat -e -c /etc/filebeat/05-nginx-es.yaml 

#3.访问nginx服务
[root@elk01:1 ~]# while true;do curl 10.0.0.211; sleep 1 ;done

浏览器访问：10.0.0.211:5601

创建索引

进入菜单栏选择Discover

查看数据（支持多条件查询）

ES的常见术语-分片

index

- index: 索引
索引是一个逻辑的存储单元，一般用于标识不同的业务类型，因此索引的名称大家可以自定义。
	用户基于索引进行数据的读写。

shard

- shard: 分片
	一个索引最少有一个分片，数据存储在分片中。
	当索引的分片数量大于1时，ES集群自动将分片分散地存储在不同的数据节点，而是实现了数据的分布式存储。

​ 一个 分片 是一个底层的 工作单元 ，它仅保存了全部数据中的一部分。 在分片内部机制中，我们将详细介绍分片是如何工作的，而现在我们只需知道一个分片是一个 Lucene 的实例，以及它本身就是一个完整的搜索引擎。 我们的文档被存储和索引到分片内，但是应用程序是直接与索引而不是与分片进行交互。
​ Elasticsearch 是利用分片将数据分发到集群内各处的。分片是数据的容器，文档保存在分片内，分片又被分配到集群内的各个节点里。 当你的集群规模扩大或者缩小时， Elasticsearch 会自动的在各节点中迁移分片，使得数据仍然均匀分布在集群里。

​ 一个分片可以是 主 分片或者 副本 分片。 索引内任意一个文档都归属于一个主分片，所以主分片的数目决定着索引能够保存的最大数据量。

replica

- replica: 副本
	每个分片有0个或多个副本，若副本数量大于0时，则可以实现对主分片的备份。为了方便理解将其分片主分片(primary shard)和副本分片(replica shard)。
	主分片(primary shard)和副本分片(replica shard)不能在同一台节点。
	主分片负责数据的读写，而副本分片只负责数据的读取和同步。
	当主分片和副本分片较多时，默认的写入策略为： "(主分片数量 + 副本分片数量) / 2 + 1"，说白了，就是半数以上写入机制。
	
	比如1个主分片，和1个副本分片总共2个分片的数据相同，其写入策略为： (1 + 1) / 2 + 1，结果为：2，表示主分片和副本分片都必须写入成功则判定成功。

​ 一个副本分片只是一个主分片的拷贝。副本分片作为硬件故障时保护数据不丢失的冗余备份，并为搜索和返回文档等读操作提供服务。
副本分片的主要目的就是为了故障转移，如果持有主分片的节点挂掉了，一个副本分片就会晋升为主分片的角色。在索引写入时，副本分片做着与主分片相同的工作。新文档首先被索引进主分片然后再同步到其它所有的副本分片。

​ 增加副本数并不会增加索引容量，在索引建立的时候就已经确定了主分片数，但是副本分片数可以随时修改

通俗理解

分片数（number_of_shards）就是主分片个数
副本数（number_of_replicas） 可以理解成主分片的拷贝个数
如果主分片数是5，则每个副本都有5个分片
因此ES集群总的分片数 = 分片数 * (副本数 + 1)

副本数分配算法：

副本数 <= ES集群的服务器个数 - 1

副本除了提升查询效率，主要目的为了容灾，数据冗余存储，保证数据安全，因此，副本数应跟据ES集群中的数据库数量（N-1）进行配置，如果ES集群中有3台服务器，则副本数设置为2，减 1 因为主分片本身也算一份。副本数设置为2，数据实际上有3份，分散到3台服务器中。

预估要存入ES总数据量 * (副本数+1) <= ES集群的总内存 / 2

存入ES搜索引擎 的数据可以预估，比如预估存入的数据有 100G，则实际上会占用的内存为 100G * (副本数+1) 加 1 因为计入了主分片，数量里超限了也可以存入，只要磁盘够大，但是计算时就可能用到了磁盘，磁盘的IO性能会让整个ES集群性能大副下降。

ES集群的总内存只取了一半，因为考虚 ES 的 java. 程序本身占用的内存，数据聚合运算消耗的内存，如果实际项目中大量复杂运算，占用的内存需要更多。

分片数分配算法：

分片数 * (副本数 + 1) <= ES集群的总CPU核心个数

分片内的检索运算聚合需要 CPU， 为了充分发挥多核CPU的性能，且又避免CPU频繁调度，因此，ES集群内总的分片数应小于等于总的CPU核心个数。

document

- document: 文档
	ES本身就是一个文档型数据库，和mangoDB类似。
	每个文件的数据分为两个部分： 元数据和源数据。
	
	1.元数据(metadata):
		用于描述源数据的数据。
		常见的字段如下:
			_index:
				表示文档隶属于哪个索引。
			_id:
				表示文档的唯一标识。
			_source:
				代表用户实际存储的数据。
	
	2.源数据(source data):
		用户实际写入的数据。
		
		- allocation(分配):
  指的是将索引的不同分片和副本分配到整个ES集群的过程，这个分配过程由ES集群自己实现。