ES (ElasticSearch) 简易解读（三）企业级日志分析ELK架构的搭建与使用

elasticsearch Logstash Kibana简称elk

架构

工作机制

• 在需要收集日志的应用上安装filebeat
• 启动filebeat后，会收集该应用的日志推送给redis，
• 然后logstash从redis中收集日志推送到Elasticsearch

Logstash

主要是用来日志的搜集、分析、过滤日志的工具

# 下载 https://www.elastic.co/cn/downloads/logstash
# 注意logstash的版本需要与elasticsearch的版本保持一致
# 安装完成后解压logstash，即可安装成功
tar -zxvf logstash...
# 在logstash目录创建一个文件夹，用来存放同步数据需要用到的东西
mkdir sync

1
2
3
4
5
6
7

官网下载 http://www.elastic.co/cn/products/logstash

---

与数据库的同步需要使用连接jar包
下载地址：https://mvnrepository.com/artifact/mysql/mysql-connector-java
选择与数据库版本一致的jar包上传到sync文件夹

---

修改配置文件

• 修改 jdbc.conf

input {
    stdin {
    }
    jdbc {
      # 连接mysql地址和库
      jdbc_connection_string => "jdbc:mysql://localhost:3306/tsdn"
      # 登录名密码
      jdbc_user => "root"
      jdbc_password => "123456"
      # 上传的mysql-connector的jar包，确定路径
      jdbc_driver_library => "/home/resources/mysql-connector-java-5.1.38-bin.jar"
      # 驱动类全路径名称
      jdbc_driver_class => "com.mysql.jdbc.Driver"
        # 开启分页，和分页最大值，自定义设置
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
        #连接数据库执行的sql语句，可以在文件中设定搜索，更新等
      statement_filepath => "/home/resources/jdbc.sql"
        #设定监听事件间隔
      schedule => "* * * * *"
      #导入es中的类型,es中有个字段是type
      type => "info_type"
    }
}
filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}
output {
    #设定收集数据的输出位置
    elasticsearch {
        #es的连接地址
        hosts => "192.168.168.101:9200"
        #新建的索引 , 数据存入时所在的索引名称
        index => "tsdn_info"
        #使用搜出出来的那个字段设定docid，我们使用商品id
        document_id => "%{id}" 
		template_overwrite => true
        template =>"/home/software/logstash5.6.0/
        #template模板文件,定义,输入到es中的数据类型的mapping
        template/template.json"
    }
    stdout {
        codec => json_lines
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

• 修改jdbc.sql

SELECT
		i.id as itemId,
		i.item_name as itemName,
		i.sell_counts as sellCounts
		i.updated_time as updated_time
FROM
		items i
WHERE
		ii.is_main = 1
		-- 以上的查询条件都不重要，只要保证数据库中的更新时间这一列与配置文件中的名字保持一致就行
		-- AND条件之后内容代表es同步数据的条件 
		-- sql_last_value是logstash中的关键字
AND
		i.updated_time >= :sql_last_value

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

配置完成后，启动logstash同步查询

# 使用logstash中bin目录下的命令，指明配置文件进行启动即可
../bin/logstash -f logstash-db-sync.conf

1
2
3

参考 Logstash 安装与使用 （https://blog.csdn.net/HueyLong/article/details/119842817）

• 后台启动logstash

 nohup ./logstash -f XXXX.conf &
1

---

Kibana

是一个开源和免费的工具，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面

Elasticsearch是集群，只需某台Elasticsearc安装一个Kibana，
需要安装一个head插件。，提供给运维查看数据状态
1
2

• 安装

# 准备安装包
kibana-5.5.2-linux-x86_64.tar.gz
#解压安装
tar -xvf kibana-5.5.2-linux-x86_64.tar.gz
#修改安装目录下的config下的kibana.yml
1
2
3
4
5

修改内容

2 server.port: 5601
7 server.host: "192.168.168.101"
21 elasticsearch.url: "http://192.168.168.101:9200"
30 kibana.index: ".kibana"
1
2
3
4

---

设置完成后，切换到bin目录，输入./kibana启动kibana（要先启动es）

• 测试访问，我的是192.168.168.101:5601