今日内容:
1) JAVA API 操作 ES 集群
2) ES的架构原理
3) ES的 sql操作
4) Beats基本概念及其使用
5) logstash基本概念及其使用
6) kibana基本概念及其使用
1) JAVA API 操作 ES 集群 : 根据关键词查询 分页查询(浅分页 和 深分页) 高亮展示数据
2) 构建索引库的时候, 除了可以指定 mapping信息以外, 还可以指定 分片和副本
PUT /job_idx_shard
{
"mappings": {
"properties": {
"id": { "type": "long", "store": true },
"area": { "type": "keyword", "store": true },
"exp": { "type": "keyword", "store": true },
"edu": { "type": "keyword", "store": true },
"salary": { "type": "keyword", "store": true },
"job_type": { "type": "keyword", "store": true },
"cmp": { "type": "keyword", "store": true },
"pv": { "type": "keyword", "store": true },
"title": { "type": "text", "store": true },
"jd": { "type": "text"}
}
},
"settings": {
"number_of_shards": 3,
"number_of_replicas": 2
}
}
3) beats基本介绍:
本质上是一个数据发送器, 用于将本地服务器中数据发生到logstash或者elasticSearch
fileBeats 主要的作用 监控本地服务器中日志数据, 将监控的数据采集下来, 然后发生到 logstash或者elasticSearch
fileBeats的组成:
input : 输入 用于监控指定的文件
output: 输出 用于将数据输出指定的目的的
4) logstash基本介绍:
数据采集工具, 主要用于将各个数据源中数据进行采集, 对数据进行处理, 将处理后数据, 发生到目的地
Flume : 数据采集的工具, 跟logstash是同类型软件 都是数据采集的工具
flume和 logstash的对比:
flume 是一款通用的采集工具 , 关注数据的传输, 更加纯粹的数据传输(采集)工具
logstash 更多时候是和beats 以及 elasticSearch组合使用 , 更多关注数据的预处理
logstash 和 fileBeats的对比:
logstash 和filebeat都具有日志收集功能，Filebeat更轻量，占用资源更少
logstash基于JVM运行, 占用资源比较大
logstash可以基于filter对数据进行过滤处理工作, 而fileBeats更多关注数据的传递工作
在生产中, 一般关于 elastic Stack的使用:
首先在需要进行采集的服务器上挂在beat组件, 然后beats组件数据数据采集到后, 发生到logstash,
然后logstash通过filter对进行处理, 处理后, 将数据传递到elasticSearch中, 后续基于kibana对
数据进行探索对数据进行图标展示
可能出现的问题:
1) 安装好fileBeats(logstash | kibana)后, 开始执行, 然后使用ctrl+C 无法退出问题: finalShell
如何解决:
ctrl +z 但是此操作可以回到命令行, 但是不会将 fileBeats关闭
此时如果直接在此启动 fileBeats 回报一个文件以及被占用错误: data path ....
处理的方式:
ps -ef | grep filebeat 找到进程ID 直接 kill -9 杀死即可
2) 路径上的问题: 在配置文件中, 路径指向不对
解决方案: 看到任何路径 都去验证一下是否正常
3) kibana无法启动问题, 即使在 kibana根目录下启动, 依然无法启动
怀疑 配置文件修改错误: kibana.yml
server.host: "node1.itcast.cn" 没有修改, 或者改了以后前面的#没删除
第二章 Beats、Logstash与Kibana
学习目标
l 能够安装部署FileBeat、Logstash和Kibana
l 能够使用FileBeat采集数据
l 能够使用Logstash采集、解析数据
l 能够使用Kibana进行数据探索和可视化
1. Beats
Beats是一个开放源代码的数据发送器。我们可以把Beats作为一种代理安装在我们的服务器上，这样就可以比较方便地将数据发送到Elasticsearch或者Logstash中。Elastic Stack提供了多种类型的Beats组件。
|
审计数据 |
AuditBeat |
|
日志文件 |
FileBeat |
|
云数据 |
FunctionBeat |
|
可用性数据 |
HeartBeat |
|
系统日志 |
JournalBeat |
|
指标数据 |
MetricBeat |
|
网络流量数据 |
PacketBeat |
|
Windows事件日志 |
Winlogbeat |
Beats可以直接将数据发送到Elasticsearch或者发送到Logstash，基于Logstash可以进一步地对数据进行处理，然后将处理后的数据存入到Elasticsearch，最后使用Kibana进行数据可视化。
1.1 FileBeat简介 阿善看到
FileBeat专门用于转发和收集日志数据的轻量级采集工具。它可以为作为代理安装在服务器上，FileBeat监视指定路径的日志文件，收集日志数据，并将收集到的日志转发到Elasticsearch或者Logstash。
1.2 FileBeat的工作原理
启动FileBeat时，会启动一个或者多个输入（Input），这些Input监控指定的日志数据位置。FileBeat会针对每一个文件启动一个Harvester（收割机）。Harvester读取每一个文件的日志，将新的日志发送到libbeat，libbeat将数据收集到一起，并将数据发送给输出（Output）。
1.3 安装FileBeat
安装FileBeat只需要将FileBeat Linux安装包上传到Linux系统，并将压缩包解压到系统就可以了。FileBeat官方下载地址：https://www.elastic.co/cn/downloads/past-releases/filebeat-7-6-1
上传FileBeat安装到Linux，并解压。
tar -xvzf filebeat-7.6.1-linux-x86_64.tar.gz
1.4 使用FileBeat采集Kafka日志到Elasticsearch
1.4.1 需求分析
在资料中有一个kafka_server.log.tar.gz压缩包，里面包含了很多的Kafka服务器日志，现在我们为了通过在Elasticsearch中快速查询这些日志，定位问题。我们需要用FileBeats将日志数据上传到Elasticsearch中。
问题：
l 首先，我们要指定FileBeat采集哪些Kafka日志，因为FileBeats中必须知道采集存放在哪儿的日志，才能进行采集。
l 其次，采集到这些数据后，还需要指定FileBeats将采集到的日志输出到Elasticsearch，那么Elasticsearch的地址也必须指定。
1.4.2 配置FileBeats
FileBeats配置文件主要分为两个部分。
- inputs
- output
从名字就能看出来，一个是用来输入数据的，一个是用来输出数据的。
1.4.2.1 input配置
|
filebeat.inputs: - type: log enabled: true paths: - /var/log/*.log #- c:programdataelasticsearchlogs* |
在FileBeats中，可以读取一个或多个数据源。
1.4.2.2 output配置
默认FileBeat会将日志数据放入到名称为：filebeat-%filebeat版本号%-yyyy.MM.dd 的索引中。
PS：
FileBeats中的filebeat.reference.yml包含了FileBeats所有支持的配置选项。
1.4.3 配置文件
- 创建配置文件
|
cd /export/server/es/filebeat-7.6.1-linux-x86_64 vim filebeat_kafka_log.yml |
- 复制一下到配置文件中
|
filebeat.inputs: - type: log enabled: true paths: - /export/server/es/data/kafka/server.log.*
output.elasticsearch: hosts: ["node1.itcast.cn:9200", "node2.itcast.cn:9200", "node3.itcast.cn:9200"] |
1.4.4 运行FileBeat
- 运行FileBeat
./filebeat -c filebeat_kafka_log.yml -e
- 将日志数据上传到/var/kafka/log，并解压
mkdir -p /export/server/es/data/kafka/
tar -xvzf kafka_server.log.tar.gz
注意: 文件权限的报错
如果在启动fileBeat的时候, 报了一个配置文件权限的错误, 请修改其权限为 -rw-r--r--
1.4.5 查询数据
- 查看索引信息
GET /_cat/indices?v
|
{ "health": "green", "status": "open", "index": "filebeat-7.6.1-2020.05.29-000001", "uuid": "dplqB_hTQq2XeSk6S4tccQ", "pri": "1", "rep": "1", "docs.count": "213780", "docs.deleted": "0", "store.size": "71.9mb", "pri.store.size": "35.8mb" } |
- 查询索引库中的数据
GET /filebeat-7.6.1-2020.05.29-000001/_search
|
{ "_index": "filebeat-7.6.1-2020.05.29-000001", "_type": "_doc", "_id": "-72pX3IBjTeClvZff0CB", "_score": 1, "_source": { "@timestamp": "2020-05-29T09:00:40.041Z", "log": { "offset": 55433, "file": { "path": "/var/kafka/log/server.log.2020-05-02-16" } }, "message": "[2020-05-02 16:01:30,682] INFO Socket connection established, initiating session, client: /192.168.88.100:46762, server: node1.itcast.cn/192.168.88.100:2181 (org.apache.zookeeper.ClientCnxn)", "input": { "type": "log" }, "ecs": { "version": "1.4.0" }, "host": { "name": "node1.itcast.cn" }, "agent": { "id": "b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64", "version": "7.6.1", "type": "filebeat", "ephemeral_id": "b8fbf7ab-bc37-46dd-86c7-fa7d74d36f63", "hostname": "node1.itcast.cn" } } } |
FileBeat自动给我们添加了一些关于日志、采集类型、Host各种字段。
1.4.6 解决一个日志涉及到多行问题
我们在日常日志的处理中，经常会碰到日志中出现异常的情况。类似下面的情况：
|
[2020-04-30 14:00:05,725] WARN [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Error when sending leader epoch request for Map(test_10m-2 -> (currentLeaderEpoch=Optional[161], leaderEpoch=158)) (kafka.server.ReplicaFetcherThread) java.io.IOException: Connection to node2.itcast.cn:9092 (id: 1 rack: null) failed. at org.apache.kafka.clients.NetworkClientUtils.awaitReady(NetworkClientUtils.java:71) at kafka.server.ReplicaFetcherBlockingSend.sendRequest(ReplicaFetcherBlockingSend.scala:102) at kafka.server.ReplicaFetcherThread.fetchEpochEndOffsets(ReplicaFetcherThread.scala:310) at kafka.server.AbstractFetcherThread.truncateToEpochEndOffsets(AbstractFetcherThread.scala:208) at kafka.server.AbstractFetcherThread.maybeTruncate(AbstractFetcherThread.scala:173) at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:113) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96) [2020-04-30 14:00:05,725] INFO [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Retrying leaderEpoch request for partition test_10m-2 as the leader reported an error: UNKNOWN_SERVER_ERROR (kafka.server.ReplicaFetcherThread) [2020-04-30 14:00:08,731] WARN [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Connection to node 1 (node2.itcast.cn/192.168.88.101:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) |
在FileBeat中，Harvest是逐行读取日志文件的。但上述的日志会出现一条日志，跨多行的情况。有异常信息时，肯定会出现多行。我们先来看一下，如果默认不处理这种情况会出现什么问题。
1.4.6.1 导入错误日志
- 在/export/server/es/data/kafka/中创建名为server.log.2020-09-10的日志文件
- 将资料中的err.txt日志文本贴入到该文件中
观察FileBeat，发现FileBeat已经针对该日志文件启动了Harvester，并读取到数据数据。
2020-05-29T19:11:01.236+0800 INFO log/harvester.go:297 Harvester started for file: /var/kafka/log/server.log.2020-09-10
- 在Elasticsearch检索该文件。
|
GET /filebeat-7.6.1-2020.05.29-000001/_search { "query": { "match": { "log.file.path": "/var/kafka/log/server.log.2020-09-10" } } } |
我们发现，原本是一条日志中的异常信息，都被作为一条单独的消息来处理了~
这明显是不符合我们的预期的，我们想要的是将所有的异常消息合并到一条日志中。那针对这种情况该如何处理呢？
1.4.6.2 问题分析
每条日志都是有统一格式的开头的，就拿Kafka的日志消息来说，[2020-04-30 14:00:05,725]这是一个统一的格式，如果不是以这样的形式开头，说明这一行肯定是属于某一条日志，而不是独立的一条日志。所以，我们可以通过日志的开头来判断某一行是否为新的一条日志。
1.4.6.3 FileBeat多行配置选项
在FileBeat的配置中，专门有一个解决一条日志跨多行问题的配置。主要为以下三个配置：
|
multiline.pattern: ^[ multiline.negate: false multiline.match: after |
multiline.pattern表示能够匹配一条日志的模式，默认配置的是以[开头的才认为是一条新的日志。
multiline.negate:配置该模式是否生效，默认为false。
multiline.match:表示是否将未匹配到的行追加到上一日志，还是追加到下一个日志。
1.4.6.4 重新配置FileBeat
- 修改filebeat.yml，并添加以下内容
|
filebeat.inputs: - type: log enabled: true paths: - /var/kafka/log/server.log.* multiline.pattern: '^[' multiline.negate: true multiline.match: after
output.elasticsearch: hosts: ["node1.itcast.cn:9200", "node2.itcast.cn:9200", "node3.itcast.cn:9200"] |
- 修改「注册表」/data.json，将server.log.2020-09-10对应的offset设置为0
cd /export/server/es/filebeat-7.6.1-linux-x86_64/data/registry/filebeat
vim data.json
- 删除之前创建的文档
|
// 删除指定文件的文档 POST /filebeat-7.6.1-2020.05.29-000001/_delete_by_query { "query": { "match": { "log.file.path": "/var/kafka/log/server.log.2020-09-10" } } } |
- 重新启动FileBeat
./filebeat -e
1.5 FileBeat是如何工作的
FileBeat主要由input和harvesters（收割机）组成。这两个组件协同工作，并将数据发送到指定的输出。
1.5.1 input和harvester
1.5.1.1 inputs（输入）
input是负责管理Harvesters和查找所有要读取的文件的组件。如果输入类型是 log，input组件会查找磁盘上与路径描述的所有文件，并为每个文件启动一个Harvester，每个输入都独立地运行。
1.5.1.2 Harvesters（收割机）
Harvesters负责读取单个文件的内容，它负责打开/关闭文件，并逐行读取每个文件的内容，并将读取到的内容发送给输出，每个文件都会启动一个Harvester。但Harvester运行时，文件将处于打开状态。如果文件在读取时，被移除或者重命名，FileBeat将继续读取该文件。
1.5.2 FileBeats如何保持文件状态
FileBeat保存每个文件的状态，并定时将状态信息保存在磁盘的「注册表」文件中，该状态记录Harvester读取的最后一次偏移量，并确保发送所有的日志数据。如果输出（Elasticsearch或者Logstash）无法访问，FileBeat会记录成功发送的最后一行，并在输出（Elasticsearch或者Logstash）可用时，继续读取文件发送数据。在运行FileBeat时，每个input的状态信息也会保存在内存中，重新启动FileBeat时，会从「注册表」文件中读取数据来重新构建状态。
在/export/server/es/filebeat-7.6.1-linux-x86_64/data目录中有一个Registry文件夹，里面有一个data.json，该文件中记录了Harvester读取日志的offset。
2. Logstash
2.1 简介
Logstash是一个开源的数据采集引擎。它可以动态地将不同来源的数据统一采集，并按照指定的数据格式进行处理后，将数据加载到其他的目的地。最开始，Logstash主要是针对日志采集，但后来Logstash开发了大量丰富的插件，所以，它可以做更多的海量数据的采集。
它可以处理各种类型的日志数据，例如：Apache的web log、Java的log4j日志数据，或者是系统、网络、防火墙的日志等等。它也可以很容易的和Elastic Stack的Beats组件整合，也可以很方便的和关系型数据库、NoSQL数据库、Kafka、RabbitMQ等整合。
2.1.1 经典架构
2.1.2 对比Flume
- Apache Flume是一个通用型的数据采集平台，它通过配置source、channel、sink来实现数据的采集，支持的平台也非常多。而Logstash结合Elastic Stack的其他组件配合使用，开发、应用都会简单很多
- Logstash比较关注数据的预处理，而Flume跟偏重数据的传输，几乎没有太多的数据解析预处理，仅仅是数据的产生，封装成Event然后传输。
2.1.3 对比FileBeat
l logstash是jvm跑的，资源消耗比较大
l 而FileBeat是基于golang编写的，功能较少但资源消耗也比较小，更轻量级
l logstash 和filebeat都具有日志收集功能，Filebeat更轻量，占用资源更少
l logstash 具有filter功能，能过滤分析日志
l 一般结构都是filebeat采集日志，然后发送到消息队列，redis，kafka中然后logstash去获取，利用filter功能过滤分析，然后存储到elasticsearch中
2.2 安装Logstash
- 切换到itcast用户
- 下载Logstash
https://www.elastic.co/cn/downloads/past-releases/logstash-7-6-1
- 解压Logstash到指定目录
- 运行测试
|
cd /export/server/es/logstash-7.6.1/ bin/logstash -e 'input { stdin { } } output { stdout {} }' |
等待一会，让Logstash启动完毕。
|
Sending Logstash logs to /export/server/es/logstash-7.6.1/logs which is now configured via log4j2.properties [2020-05-28T16:31:44,159][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified [2020-05-28T16:31:44,264][INFO ][logstash.runner ] Starting Logstash {"logstash.version"=>"7.6.1"} [2020-05-28T16:31:45,631][INFO ][org.reflections.Reflections] Reflections took 37 ms to scan 1 urls, producing 20 keys and 40 values [2020-05-28T16:31:46,532][WARN ][org.logstash.instrument.metrics.gauge.LazyDelegatingGauge][main] A gauge metric of an unknown type (org.jruby.RubyArray) has been create for key: cluster_uuids. This may result in invalid serialization. It is recommended to log an issue to the responsible developer/development team. [2020-05-28T16:31:46,560][INFO ][logstash.javapipeline ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>250, "pipeline.sources"=>["config string"], :thread=>"#<Thread:0x3ccbc15b run>"} [2020-05-28T16:31:47,268][INFO ][logstash.javapipeline ][main] Pipeline started {"pipeline.id"=>"main"} The stdin plugin is now waiting for input: [2020-05-28T16:31:47,348][INFO ][logstash.agent ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]} [2020-05-28T16:31:47,550][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600} |
然后，随便在控制台中输入内容，等待Logstash的输出。
|
{ "host" => "node1.itcast.cn", "message" => "hello world", "@version" => "1", "@timestamp" => 2020-05-28T08:32:31.007Z } |
ps：
-e选项表示，直接把配置放在命令中，这样可以有效快速进行测试
2.3 采集Apache Web服务器日志
2.3.1 需求
Apache的Web Server会产生大量日志，当我们想要对这些日志检索分析。就需要先把这些日志导入到Elasticsearch中。此处，我们就可以使用Logstash来实现日志的采集。
打开这个文件，如下图所示。我们发现，是一个纯文本格式的日志。如下图所示：
这个日志其实由一个个的字段拼接而成，参考以下表格。
|
字段名 |
说明 |
|
client IP |
浏览器端IP |
|
timestamp |
请求的时间戳 |
|
method |
请求方式（GET/POST） |
|
uri |
请求的链接地址 |
|
status |
服务器端响应状态 |
|
length |
响应的数据长度 |
|
reference |
从哪个URL跳转而来 |
|
browser |
浏览器 |
因为最终我们需要将这些日志数据存储在Elasticsearch中，而Elasticsearch是有模式（schema）的，而不是一个大文本存储所有的消息，而是需要将字段一个个的保存在Elasticsearch中。所以，我们需要在Logstash中，提前将数据解析好，将日志文本行解析成一个个的字段，然后再将字段保存到Elasticsearch中。
2.3.2 准备日志数据
将Apache服务器日志上传到 /export/server/es/data/apache/ 目录
mkdir -p /export/server/es/data/apache/
2.3.3 使用FileBeats将日志发送到Logstash
在使用Logstash进行数据解析之前，我们需要使用FileBeat将采集到的数据发送到Logstash。之前，我们使用的FileBeat是通过FileBeat的Harvester组件监控日志文件，然后将日志以一定的格式保存到Elasticsearch中，而现在我们需要配置FileBeats将数据发送到Logstash。FileBeat这一端配置以下即可：
|
#----------------------------- Logstash output --------------------------------- #output.logstash: # Boolean flag to enable or disable the output module. #enabled: true
# The Logstash hosts #hosts: ["localhost:5044"] |
hosts配置的是Logstash监听的IP地址/机器名以及端口号。
114.113.220.255
准备FileBeat配置文件
cd /export/server/es/filebeat-7.6.1-linux-x86_64
vim filebeat-logstash.yml
因为Apache的web log日志都是以IP地址开头的，所以我们需要修改下匹配字段。
|
filebeat.inputs: - type: log enabled: true paths: - /var/apache/log/access.* multiline.pattern: '^d+.d+.d+.d+ ' multiline.negate: true multiline.match: after
output.logstash: enabled: true hosts: ["node1.itcast.cn:5044"] |
启动FileBeat，并指定使用新的配置文件
./filebeat -e -c filebeat-logstash.yml
FileBeat将尝试建立与Logstash监听的IP和端口号进行连接。但此时，我们并没有开启并配置Logstash，所以FileBeat是无法连接到Logstash的。
|
2020-06-01T11:28:47.585+0800 ERROR pipeline/output.go:100 Failed to connect to backoff(async(tcp://node1.itcast.cn:5044)): dial tcp 192.168.88.100:5044: connect: connection refused |
2.3.4 配置Logstash接收FileBeat数据并打印
Logstash的配置文件和FileBeat类似，它也需要有一个input、和output。基本格式如下：
|
# #号表示添加注释 # input表示要接收的数据 input { }
# file表示对接收到的数据进行过滤处理 filter {
}
# output表示将数据输出到其他位置 output { } |
配置从FileBeat接收数据
cd /export/server/es/logstash-7.6.1/config
vim filebeat-print.conf
|
input { beats { port => 5044 } }
output { stdout { codec => rubydebug } } |
测试logstash配置是否正确
bin/logstash -f config/filebeat-print.conf --config.test_and_exit
|
[2020-06-01T11:46:33,940][INFO ][logstash.runner ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash |
启动logstash
bin/logstash -f config/filebeat-print.conf --config.reload.automatic
reload.automatic：修改配置文件时自动重新加载
测试
创建一个access.log.1文件，使用cat test >> access.log.1往日志文件中追加内容。
test文件中只保存一条日志：
|
[root@node1 log]# cat test 235.9.200.242 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249 |
当我们启动Logstash之后，就可以发现Logstash会打印出来从FileBeat接收到的数据：
|
{ "log" => { "file" => { "path" => "/var/apache/log/access.log.1" }, "offset" => 825 }, "input" => { "type" => "log" }, "agent" => { "ephemeral_id" => "d4c3b652-4533-4ebf-81f9-a0b78c0d4b05", "version" => "7.6.1", "type" => "filebeat", "id" => "b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64", "hostname" => "node1.itcast.cn" }, "@timestamp" => 2020-06-01T09:07:55.236Z, "ecs" => { "version" => "1.4.0" }, "host" => { "name" => "node1.itcast.cn" }, "tags" => [ [0] "beats_input_codec_plain_applied" ], "message" => "235.9.200.242 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249", "@version" => "1" } |
2.3.5 Logstash输出数据到Elasticsearch
通过控制台，我们发现Logstash input接收到的数据没有经过任何处理就发送给了output组件。而其实我们需要将数据输出到Elasticsearch。所以，我们修改Logstash的output配置。配置输出Elasticsearch只需要配置以下就可以了：
|
output { elasticsearch { hosts => [ "localhost:9200" ] }} |
操作步骤：
- 重新拷贝一份配置文件
cp filebeat-print.conf filebeat-es.conf
- 将output修改为Elasticsearch
|
input { beats { port => 5044 } }
output { elasticsearch { hosts => [ "node1.itcast.cn:9200","node2.itcast.cn:9200","node3.itcast.cn:9200"] } } |
- 重新启动Logstash
bin/logstash -f config/filebeat-es.conf --config.reload.automatic
- 追加一条日志到监控的文件中，并查看Elasticsearch中的索引、文档
cat test >> access.log.1
// 查看索引数据
GET /_cat/indices?v
我们在Elasticsearch中发现一个以logstash开头的索引。
|
{ "health": "green", "status": "open", "index": "logstash-2020.06.01-000001", "uuid": "147Uwl1LRb-HMFERUyNEBw", "pri": "1", "rep": "1", "docs.count": "2", "docs.deleted": "0", "store.size": "44.8kb", "pri.store.size": "22.4kb" } |
// 查看索引库的数据
GET /logstash-2020.06.01-000001/_search?format=txt
{
"from": 0,
"size": 1
}
我们可以获取到以下数据：
|
"@timestamp": "2020-06-01T09:38:00.402Z", "tags": [ "beats_input_codec_plain_applied" ], "host": { "name": "node1.itcast.cn" }, "@version": "1", "log": { "file": { "path": "/var/apache/log/access.log.1" }, "offset": 1343 }, "agent": { "version": "7.6.1", "ephemeral_id": "d4c3b652-4533-4ebf-81f9-a0b78c0d4b05", "id": "b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64", "hostname": "node1.itcast.cn", "type": "filebeat" }, "input": { "type": "log" }, "message": "235.9.200.242 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249", "ecs": { "version": "1.4.0" }
|
从输出返回结果，我们可以看到，日志确实已经保存到了Elasticsearch中，而且我们看到消息数据是封装在名为message中的，其他的数据也封装在一个个的字段中。我们其实更想要把消息解析成一个个的字段。例如：IP字段、时间、请求方式、请求URL、响应结果，这样。
2.3.6 Logstash过滤器
在Logstash中可以配置过滤器Filter对采集到的数据进行中间处理，在Logstash中，有大量的插件供我们使用。参考官网：
https://www.elastic.co/guide/en/logstash/7.6/filter-plugins.html
此处，我们重点来讲解Grok插件。
2.3.6.1 查看Logstash已经安装的插件
bin/logstash-plugin list
2.3.6.2 Grok插件
Grok是一种将非结构化日志解析为结构化的插件。这个工具非常适合用来解析系统日志、Web服务器日志、MySQL或者是任意其他的日志格式。
Grok官网：https://www.elastic.co/guide/en/logstash/7.6/plugins-filters-grok.html
2.3.6.3 Grok语法
Grok是通过模式匹配的方式来识别日志中的数据,可以把Grok插件简单理解为升级版本的正则表达式。它拥有更多的模式，默认，Logstash拥有120个模式。如果这些模式不满足我们解析日志的需求，我们可以直接使用正则表达式来进行匹配。
官网：https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
grok模式的语法是：%{SYNTAX:SEMANTIC}
SYNTAX指的是Grok模式名称，SEMANTIC是给模式匹配到的文本字段名。例如：
|
%{NUMBER:duration} %{IP:client} duration表示：匹配一个数字，client表示匹配一个IP地址。 |
默认在Grok中，所有匹配到的的数据类型都是字符串，如果要转换成int类型（目前只支持int和float），可以这样：%{NUMBER:duration:int} %{IP:client}
以下是常用的Grok模式：
|
NUMBER |
匹配数字（包含：小数） |
|
INT |
匹配整形数字 |
|
POSINT |
匹配正整数 |
|
WORD |
匹配单词 |
|
DATA |
匹配所有字符 |
|
IP |
匹配IP地址 |
|
PATH |
匹配路径 |
2.3.6.4 用法
|
filter { grok { match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" } } } |
2.3.7 匹配日志中的IP、日期并打印
|
235.9.200.242 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249 |
我们使用IP就可以把前面的IP字段匹配出来，使用HTTPDATE可以将后面的日期匹配出来。
配置Grok过滤插件
- 配置Logstash
|
input { beats { port => 5044 } }
filter { grok { match => { "message" => "%{IP:ip} - - [%{HTTPDATE:date}]" } } }
output { stdout { codec => rubydebug } } |
- 启动Logstash
bin/logstash -f config/filebeat-filter-print.conf --config.reload.automatic
|
{ "log" => { "offset" => 1861, "file" => { "path" => "/var/apache/log/access.log.1" } }, "input" => { "type" => "log" }, "tags" => [ [0] "beats_input_codec_plain_applied" ], "date" => "15/Apr/2015:00:27:19 +0849", "ecs" => { "version" => "1.4.0" }, "@timestamp" => 2020-06-01T11:02:05.809Z, "message" => "235.9.200.242 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249", "host" => { "name" => "node1.itcast.cn" }, "ip" => "235.9.200.242", "agent" => { "hostname" => "node1.itcast.cn", "version" => "7.6.1", "ephemeral_id" => "d4c3b652-4533-4ebf-81f9-a0b78c0d4b05", "id" => "b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64", "type" => "filebeat" }, "@version" => "1" } |
我们看到，经过Grok过滤器插件处理之后，我们已经获取到了ip和date两个字段。接下来，我们就可以继续解析其他的字段。
2.3.8 解析所有字段
将日志解析成以下字段：
|
字段名 |
说明 |
|
client IP |
浏览器端IP |
|
timestamp |
请求的时间戳 |
|
method |
请求方式（GET/POST） |
|
uri |
请求的链接地址 |
|
status |
服务器端响应状态 |
|
length |
响应的数据长度 |
|
reference |
从哪个URL跳转而来 |
|
browser |
浏览器 |
- 修改Logstash配置文件
|
input { beats { port => 5044 } }
filter { grok { match => { "message" => "%{IP:ip} - - [%{HTTPDATE:date}] "%{WORD:method} %{PATH:uri} %{DATA}" %{INT:status} %{INT:length} "%{DATA:reference}" "%{DATA:browser}"" } } }
output { stdout { codec => rubydebug } } |
- 测试并启动Logstash
我们可以看到，8个字段都已经成功解析。
|
{ "reference" => "www.baidu.com", "@version" => "1", "ecs" => { "version" => "1.4.0" }, "@timestamp" => 2020-06-02T03:30:10.048Z, "ip" => "235.9.200.241", "method" => "POST", "uri" => "/itcast.cn/bigdata.html", "agent" => { "id" => "b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64", "ephemeral_id" => "734ae9d8-bcdc-4be6-8f97-34387fcde972", "version" => "7.6.1", "hostname" => "node1.itcast.cn", "type" => "filebeat" }, "length" => "45", "status" => "200", "log" => { "file" => { "path" => "/var/apache/log/access.log" }, "offset" => 1 }, "input" => { "type" => "log" }, "host" => { "name" => "node1.itcast.cn" }, "tags" => [ [0] "beats_input_codec_plain_applied" ], "browser" => "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900", "date" => "15/Apr/2015:00:27:19 +0849", "message" => "235.9.200.241 - - [15/Apr/2015:00:27:19 +0849] "POST /itcast.cn/bigdata.html HTTP/1.1" 200 45 "www.baidu.com" "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900"" } |
2.3.9 将数据输出到Elasticsearch
到目前为止，我们已经通过了Grok Filter可以将日志消息解析成一个一个的字段，那现在我们需要将这些字段保存到Elasticsearch中。我们看到了Logstash的输出中，有大量的字段，但如果我们只需要保存我们需要的8个，该如何处理呢？而且，如果我们需要将日期的格式进行转换，我们又该如何处理呢？
2.3.9.1 过滤出来需要的字段
要过滤出来我们需要的字段。我们需要使用mutate插件。mutate插件主要是作用在字段上，例如：它可以对字段进行重命名、删除、替换或者修改结构。
官方文档：https://www.elastic.co/guide/en/logstash/7.6/plugins-filters-mutate.html
例如，mutate插件可以支持以下常用操作
配置：
注意：此处为了方便进行类型的处理，将status、length指定为int类型。
|
input { beats { port => 5044 } }
filter { grok { match => { "message" => "%{IP:ip} - - [%{HTTPDATE:date}] "%{WORD:method} %{PATH:uri} %{DATA}" %{INT:status:int} %{INT:length:int} "%{DATA:reference}" "%{DATA:browser}"" } } mutate { enable_metric => "false" remove_field => ["message", "log", "tags", "@timestamp", "input", "agent", "host", "ecs", "@version"] } }
output { stdout { codec => rubydebug } } |
2.3.9.2 转换日期格式
要将日期格式进行转换，我们可以使用Date插件来实现。该插件专门用来解析字段中的日期，官方说明文档：https://www.elastic.co/guide/en/logstash/7.6/plugins-filters-date.html
用法如下：
将date字段转换为「年月日 时分秒」格式。默认字段经过date插件处理后，会输出到@timestamp字段，所以，我们可以通过修改target属性来重新定义输出字段。
Logstash配置修改为如下：
|
input { beats { port => 5044 } }
filter { grok { match => { "message" => "%{IP:ip} - - [%{HTTPDATE:date}] "%{WORD:method} %{PATH:uri} %{DATA}" %{INT:status:int} %{INT:length:int} "%{DATA:reference}" "%{DATA:browser}"" } } mutate { enable_metric => "false" remove_field => ["message", "log", "tags", "@timestamp", "input", "agent", "host", "ecs", "@version"] } date { match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"] target => "date" } }
output { stdout { codec => rubydebug } } |
启动Logstash：
bin/logstash -f config/filebeat-filter-print.conf --config.reload.automatic
|
{ "status" => "200", "reference" => "www.baidu.com", "method" => "POST", "browser" => "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900", "ip" => "235.9.200.241", "length" => "45", "uri" => "/itcast.cn/bigdata.html", "date" => 2015-04-14T15:38:19.000Z } |
2.3.9.3 输出到Elasticsearch指定索引
我们可以通过
elasticsearch {
hosts => ["node1.itcast.cn:9200" ,"node2.itcast.cn:9200" ,"node3.itcast.cn:9200"]
index => "xxx"
}
index来指定索引名称，默认输出的index名称为：logstash-%{+yyyy.MM.dd}。但注意，要在index中使用时间格式化，filter的输出必须包含 @timestamp字段，否则将无法解析日期。
|
input { beats { port => 5044 } }
filter { grok { match => { "message" => "%{IP:ip} - - [%{HTTPDATE:date}] "%{WORD:method} %{PATH:uri} %{DATA}" %{INT:status:int} %{INT:length:int} "%{DATA:reference}" "%{DATA:browser}"" } } mutate { enable_metric => "false" remove_field => ["message", "log", "tags", "input", "agent", "host", "ecs", "@version"] } date { match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"] target => "date" } }
output { stdout { codec => rubydebug } elasticsearch { hosts => ["node1.itcast.cn:9200" ,"node2.itcast.cn:9200" ,"node3.itcast.cn:9200"] index => "apache_web_log_%{+YYYY-MM}" } } |
启动Logstash
bin/logstash -f config/filebeat-apache-weblog.conf --config.test_and_exit
bin/logstash -f config/filebeat-apache-weblog.conf --config.reload.automatic
注意：
l index名称中，不能出现大写字符
3. Kibana
3.1 简介
通过上面的这张图就可以看到，Kibana可以用来展示丰富的图表。
l Kibana是一个开源的数据分析和可视化平台，使用Kibana可以用来搜索Elasticsearch中的数据，构建漂亮的可视化图形、以及制作一些好看的仪表盘
l Kibana是用来管理Elastic stack组件的可视化平台。例如：使用Kibana可以进行一些安全设置、用户角色设置、对Elasticsearch进行快照等等
l Kibana提供统一的访问入口，不管是日志分析、还是查找文档，Kibana提供了一个使用这些功能的统一访问入口
l Kibana使用的是Elasticsearch数据源，Elasticsearch是存储和处理数据的引擎，而Kibana就是基于Elasticsearch之上的可视化平台
主要功能：
l 探索和查询Elasticsearch中的数据
l 可视化与分析
3.2 安装Kibana
在Linux下安装Kibana，可以使用Elastic stack提供 tar.gz压缩包。官方下载地址：
https://www.elastic.co/cn/downloads/past-releases/kibana-7-6-1
- 解压Kibana gz压缩包
tar -xzf kibana-7.6.1-linux-x86_64.tar.gz
- 进入到Kibana目录
cd kibana-7.6.1-linux-x86_64/
- 配置Kibana: config/kibana.yml
|
server.host: "node1.itcast.cn" # The Kibana server's name. This is used for display purposes. server.name: "itcast-kibana"
# The URLs of the Elasticsearch instances to use for all your queries. elasticsearch.hosts: ["http://node1.itcast.cn:9200","http://node2.itcast.cn:9200","http://node3.itcast.cn:9200"] |
- 运行Kibana
./bin/kibana
3.2.1 查看Kibana状态
输入以下网址，可以查看到Kibana的运行状态：
http://node1.itcast.cn:5601/status
3.2.2 查看Elasticsearch的状态
点击 
按钮，再点击 「Index Management」，可以查看到Elasticsearch集群中的索引状态。
点击索引的名字，可以进一步查看索引更多的信息。
点击「Manage」按钮，还可以用来管理索引。
3.3 添加Elasticsearch数据源
3.3.1 Kibana索引模式
在开始使用Kibana之前，我们需要指定想要对哪些Elasticsearch索引进行处理、分析。在Kibana中，可以通过定义索引模式（Index Patterns）来对应匹配Elasticsearch索引。在第一次访问Kibana的时候，系统会提示我们定义一个索引模式。或者我们可以通过点击按钮，再点击Kibana下方的Index Patterns，来创建索引模式。参考下图：
- 定义索引模式，用于匹配哪些Elasticsearch中的索引。点击「Next step」
- 选择用于进行时间过滤的字段
- 点击「Create Index Pattern」按钮，创建索引模式。创建索引模式成功后，可以看到显示了该索引模式对应的字段。里面描述了哪些可以用于搜索、哪些可以用来进行聚合计算等。
3.4 探索数据（Discovery）
通过Kibana中的Discovery组件，我们可以快速地进行数据的检索、查询。
3.4.1 使用探索数据功能
点击按钮可以打开Discovery页面。
我们发现没有展示任何的数据。但我们之前已经把数据导入到Elasticsearch中了。
Kibana提示，让我们扩大我们的查询的时间范围。
默认Kibana是展示最近15分钟的数据。我们把时间范围调得更长一些，就可以看到数据了。
将时间范围选择为1年范围内的，我们就可以查看到Elasticsearch中的数据了。
3.4.2 导入更多的Apache Web日志数据
- 将资料中的 access.log 文件上传到Linux
- 将access.log移动到/var/apache/log，并重命名为access.log.2
mv access.log /var/apache/log/access.log.2
- 启动FileBeat
./filebeat -e -c filebeat-logstash.yml
- 启动Logstash
bin/logstash -f config/filebeat-es.conf --config.reload.automatic
3.4.3 基于时间过滤查询
3.4.3.1 选择时间范围
3.4.3.2 指定查询某天的数据
查询2020年5月6日的所有日志数据。
3.4.3.3 从直方图
上选择日期更细粒度范围
如果要选择查看某一天的日志，上面这种方式会有一些麻烦，我们有更快更容易的方式。
3.4.4 使用Kibana搜索数据
在Kibana的Discovery组件中，可以在查询栏中输入搜索条件。默认情况下，可以使用Kibana内置的标准查询语言，来进行快速查询。还有一种是遗留的基于Lucene的查询语法目前暂时可用，这种查询语法也可以使用基于JSON的Elasticsearch DSL也是可用的。当我们在Discovery搜索数据时，对应的直方图、文档列表都会随即更新。默认情况下，优先展示最新的文档，按照时间倒序排序的。
3.4.4.1 Kibana查询语言（KQL）
在7.0中，Kibana上线了新的查询语言。这种语言简洁、易用，有利于快速查询。
查询语法：
「字段:值」，如果值是字符串，可以用双引号括起来。
查询包含zhihu的请求
*zhihu*
查询页面不存在的请求
status : 404
查询请求成功和不存在的请求
status: (404 or 200)
查询方式为POST请求，并请求成功的日志
status: 200 and method: post
查询方式为GET成功的请求，并且响应数据大于512的日志
status: 200 and method: get and length > 512
查询请求成功的且URL为「/itcast.cn」开头的日志
uri: "/itcast.cn/*"
注意：因为/为特殊字符，需要使用反斜杠进行转义
3.4.4.2 过滤字段
Kibana的Discovery组件提供各种各样的筛选器，这样可以筛选出来我们关注的数据上。例如：我们只想查询404的请求URI。
指定过滤出来404以及请求的URI、从哪儿跳转来的日志
将查询保存下来，方便下次直接查看
下次直接点击Open就可以直接打开之前保存的日志了
3.5 数据可视化（Visualize）
Kibana中的Visualize可以基于Elasticsearch中的索引进行数据可视化，然后将这些可视化图表添加到仪表盘中。
3.5.1 数据可视化的类型
l Lens
n 通过简单地拖拽数据字段，快速构建基本的可视化
l 常用的可视化对象
n 线形图（Line）、面积图（Area）、条形图（Bar）：可以用这些带X/Y坐标的图形来进行不同分类的比较
n 饼图（Pie）：可以用饼图来展示占比
n 数据表（Data Table）：以数据表格的形式展示
n 指标（Metrics）：以数字的方式展示
n 目标和进度：显示带有进度指标的数字
n 标签云/文字云（Tag Cloud）：以文字云方式展示标签，文字的大小与其重要性相关
l Timelion
n 从多个时间序列数据集来展示数据
l 地图
n 展示地理位置数据
l 热图
n 在矩阵的单元格展示数据
l 仪表盘工具
n Markdown部件：显示一些MD格式的说明
n 控件：在仪表盘中添加一些可以用来交互的组件
l Vega
3.5.2 以饼图展示404与200的占比
效果图：
操作步骤：
- 创建可视化
- 选择要进行可视化图形类型，此处我们选择Pie（饼图类型）
- 选择数据源
- 添加分桶、分片（还记得吗？我们在Elasticsearch进行分组聚合都是以分桶方式进行的，可以把它理解为分组）
- 配置分桶以及指标计算方式
- 点击蓝色播放按钮执行。
- 保存图形（取名为：apache_log@404_200）
3.5.3 以条形图方式展示2020年5月每日请求数
效果如下：
开发步骤：
我们还可以修改图形的样式，例如：以曲线、面积图的方式展示。
3.5.4 以TSVB可视化不同访问来源的数据
TSVB是一个基于时间序列的数据可视化工具，它可以使用Elasticsearch聚合的所有功能。使用TSVB，我们可以轻松地完成任意聚合方式来展示复杂的数据。它可以让我们快速制作效果的图表：
- 基于时间序列的图形展示
- 展示指标数据
- TopN
- 类似油量表的展示
- Markdown自定义数据展示
- 以表格方式展示数据
操作步骤：
- 创建TSVB可视化对象
- 配置Time Series数据源分组条件
- 配置Metric
- TopN
3.5.5 制作用户选择请求方式、响应字节大小控制组件
3.5.5.1 控制组件
在Kibana中，我们可以使用控件来控制图表的展示。例如：提供一个下列列表，供查看图表的用户只展示比较关注的数据。我们可以添加两个类型的控制组件：
- 选项列表
l 根据一个或多个指定选项来筛选内容。例如：我们先筛选某个城市的数据，就可以通过选项列表来选择该城市
- 范围选择滑块
l 筛选出来指定范围的数据。例如：我们筛选某个价格区间的商品等。
3.5.5.2 Kibana开发
3.6 制作Dashboard
接下来，我们把前面的几个图形放到一个看板中。这样，我们就可以在一个看板中，浏览各类数据了。
- 点击第三个组件图标，并创建一个新的Dashboard。
- 点击Edit编辑Dashboard。
- 依次添加我们之前制作好的图表。
