CentOS集成ELK，安装Logstash插件及配置Mysql数据与ES数据同步(包含多数据源设置)超详细

编程入门行业动态更新时间:2024-10-05 01:22:23

CentOS集成ELK，安装Logstash<a href=https://www.elefans.com/category/jswz/34/1771202.html style= 插件及配置Mysql数据与ES数据同步(包含多数据源设置)超详细"/>

CentOS集成ELK，安装Logstash插件及配置Mysql数据与ES数据同步(包含多数据源设置)超详细

CentOS集成ELK，安装Logstash插件及配置Mysql数据与ES数据同步（包含多数据源设置）

1、开始安装Logstash插件
2、安装gem
3、进入logstash目录，修改Gemfile
4、安装Logstash的相关插件
5、建立Logstash的同步文件
6、启动Logstash
7、停止logstash
8、设置logstash开机自启

本章内容只讲解Logstash的搭建，需要搭建Elasticsearch与Kibana服务请跳转至Elasticsearch与Kibana服务搭建

1、开始安装Logstash插件

采用yum命令安装（不推荐，使用yum有可能下载失败）
- 由于yum中是没有带有Logstash安装源的，我们需要自己添加
- 首先我们需要下载并安装公共签名密钥
  rpm --import
- 创建我们的yum关于Logstash的源文件
  vim /etc/yum.repos.d/logstash.rep
- 请将下面的内容复制到文本中，目前版本最新为7，本文章创建时间为2020-02-27日，如果版本升级需要将[logstash-7.x]中的7改为新版本号
  [logstash7-.x]
  name=Elastic repository for 7.x packages
  baseurl=.x/yum
  gpgcheck=1
  gpgkey=
  enabled=1
  autorefresh=1
  type=rpm-md
- 开始安装logstash，如果发现镜像下载太慢，请采用手动安装
  yum -y install logstash

采用RPM方式安装
- Logstash与其他相关插件地址下载
  这里我们选择最新版本-7.5.2，下载完成后将文件拷入到我们的linux主机中
- 运行我们的logstash安装包，即可完成安装
  rpm -ivh logstash-7.5.2.rpm

采用压缩包TAR形式安装（推荐）
- 解压压缩包 tar -xvf logstash-7.5.2.tar.gz
至此Logstash安装步骤已经完成

2、安装gem

yum install -y gem
设置源
添加清华镜像源并移除默认源（其他国内镜像已经用不了了，比如taobao，ruby-china，aliyuncs）
gem sources --add / --remove /
列出已有源，一般情况下只会显示一个，如果有多的请删除
gem sources -l

3、进入logstash目录，修改Gemfile

两者只是进入地址上的不同。

采用yum与RPM方式安装的请往这里看
- vim /usr/share/logstash/Gemfile #
  将sources至修改为/
  source “/”
  保存并退出

采用压缩包形式安装的请往这里看
- vim /opt/logstash-7.5.2/Gemfile #
  将sources至修改为/
  source “/”
  保存并退出

4、安装Logstash的相关插件

两者只是进入地址上的不同。

采用yum与RPM方式安装的请往这里看

进入Logstash目录
cd /usr/share/logstash
安装logstash-input-jdbc插件
bin/logstash-plugin install logstash-input-jdbc
安装logstash-output-elasticsearch插件
bin/logstash-plugin install logstash-output-elasticsearch

采用压缩包形式安装的请往这里看

进入Logstash目录
cd /opt/logstash-7.5.2
安装logstash-input-jdbc插件
bin/logstash-plugin install logstash-input-jdbc
安装logstash-output-elasticsearch插件
bin/logstash-plugin install logstash-output-elasticsearch

5、建立Logstash的同步文件

我们需要放入Mysql驱动（jar包），放入位置随意，我这放在了Logstash的文件目录下，使用的版本为
mysql-connector-java-8.0.18.jar

采用yum与RPM方式安装的请往这里看
- 我们在Logstash目录下新建一个文件夹，命名为：mysql_config
```
mkdir /etc/logstash/mysql_config
```

采用压缩包形式安装的请往这里看
- 我们在Logstash目录下新建一个文件夹，命名为：mysql_config
```
mkdir /opt/logstash-7.5.2/mysql_config
```

配置文件说明

jdbc_driver_library: jdbc mysql 驱动的路径
jdbc_driver_class: 驱动类的名字，mysql 填 com.mysql.jdbc.Driver 或 com.mysql.cj.jdbc.Driver
jdbc_connection_string: mysql 地址
jdbc_user: mysql 用户名
jdbc_password: mysql 密码
schedule: 执行 sql 时机，类似 crontab 的调度--------------->>>>>>在线crontab表达转换网址
statement: 要执行的 sql语句，以 “:” 开头是定义的变量，可以通过 parameters 来设置变量，这里的 sql_last_value 是内置的变量，表示上一次 sql 执行中 update_time 的值，这里 update_time 条件是 >= 因为时间有可能相等，没有等号可能会漏掉一些增量，如果SQL语句较长可指定SQL文件
use_column_value: 使用递增列的值
statement_filepath：是基于文件形式存储的SQL。这里存放的是路径
tracking_column_type: 递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型
tracking_column: 递增字段的名称，这里使用 update_time 这一列，这列的类型是 timestamp
last_run_metadata_path: 同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以手动修改
jdbc_paging_enabled：是否开启分页查询
jdbc_page_size：每次分页多少条数据
type：为多数据表/源做准备，作为区分条件
新建ES与Mysql同步文件
- 采用yum与RPM方式安装的请往这里看
```
vim /etc/logstash/mysql_config/table_1.conf
```
- 采用压缩包形式安装的请往这里看
```
vim /opt/logstash-7.5.2/mysql_config/table_1.conf
```

添加文本内容

input {stdin {}jdbc {#mysql 驱动的路径jdbc_driver_library => "/etc/logstash/mysql-connector-java-8.0.18.jar"#驱动类的名字jdbc_driver_class => "com.mysql.cj.jdbc.Driver"#mysql 地址，这里重点说个问题，这个时间，一定要校准，serverTimezone=UCT或者serverTimezone=Asia/Shanghai一定要校准，否则同步数据会产生错误jdbc_connection_string => "jdbc:mysql://192.168.114.136/speed?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UCT"#时区校准jdbc_default_timezone =>"Asia/Shanghai"#mysql 用户jdbc_user => "root"#mysql 密码jdbc_password => "root"#执行 sql 时机，类似 crontab 的调度，每一分钟执行一次schedule => "* * * * *"#要执行的 sql语句，以 “:” 开头是定义的变量，sql_last_value 是内置的变量，表示上一次 sql 执行中 update_time 的值，这里的字段名，根据数据库的配置，是存在大小写敏感的statement => "select * from user where UPDATE_TIME >= :sql_last_value"#是否使用递增列的值use_column_value => true#递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型tracking_column_type => "timestamp"#使用递增列的值，这里的字段名大小写不敏感tracking_column => "update_time"#同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以自定义last_run_metadata_path => "/etc/logstash/metadata/user_table"#是否分页查询jdbc_paging_enabled => "true"#每次分页的数量jdbc_page_size => "100"#为多数据表/源做准备，作为区分条件type => "type_user"}
}
output {elasticsearch {# ES的IP地址及端口hosts => ["192.168.114.136:9200"]# 索引名称 可自定义index => "user"# 需要关联的数据库中有有一个id字段，对应类型中的iddocument_id => "%{user_id}"# 文档类型（不建意使用，因为elasticsearch 7.0.0以后将不建议使用，8.0.0 以后完全不支持。我是直接删除了）document_type => "user"}stdout {# 以JSON格式输出至elasticsearchcodec => json_lines}
}

保存并退出。

多表/多数据源写法

这里含有多个JDBC，然后再下方使用if进行判断，区分ES存储

input {stdin {}jdbc {#mysql 驱动的路径jdbc_driver_library => "/etc/logstash/mysql-connector-java-8.0.18.jar"#驱动类的名字jdbc_driver_class => "com.mysql.cj.jdbc.Driver"#mysql 地址，这里重点说个问题，这个时间，一定要校准，serverTimezone=UCT或者serverTimezone=Asia/Shanghai一定要校准，否则同步数据会产生错误jdbc_connection_string => "jdbc:mysql://192.168.114.136/speed?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UCT"#时区校准jdbc_default_timezone =>"Asia/Shanghai"#mysql 用户jdbc_user => "root"#mysql 密码jdbc_password => "root"#执行 sql 时机，类似 crontab 的调度，每一分钟执行一次schedule => "* * * * *"#要执行的 sql语句，以 “:” 开头是定义的变量，sql_last_value 是内置的变量，表示上一次 sql 执行中 update_time 的值，这里的字段名，根据数据库的配置，是存在大小写敏感的statement => "select * from user where UPDATE_TIME >= :sql_last_value"#是否使用递增列的值use_column_value => true#递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型tracking_column_type => "timestamp"#使用递增列的值，这里的字段名大小写不敏感tracking_column => "update_time"#同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以自定义last_run_metadata_path => "/opt/metadata/product_table"#是否分页查询jdbc_paging_enabled => "true"#每次分页的数量jdbc_page_size => "100"#为多数据表/源做准备，作为区分条件type => "type_user"}jdbc {#mysql 驱动的路径jdbc_driver_library => "/etc/logstash/mysql-connector-java-8.0.18.jar"#驱动类的名字jdbc_driver_class => "com.mysql.cj.jdbc.Driver"#mysql 地址，这里重点说个问题，这个时间，一定要校准，serverTimezone=UCT或者serverTimezone=Asia/Shanghai一定要校准，否则同步数据会产生错误jdbc_connection_string => "jdbc:mysql://192.168.114.136/speed?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UCT"#时区校准jdbc_default_timezone =>"Asia/Shanghai"#mysql 用户jdbc_user => "root"#mysql 密码jdbc_password => "root"#执行 sql 时机，类似 crontab 的调度，每一分钟执行一次schedule => "* * * * *"#要执行的 sql语句，以 “:” 开头是定义的变量，sql_last_value 是内置的变量，表示上一次 sql 执行中 update_time 的值，这里的字段名，根据数据库的配置，是存在大小写敏感的statement => "select * from expert where UPDATE_TIME >= :sql_last_value"#是否使用递增列的值use_column_value => true#递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型tracking_column_type => "timestamp"#使用递增列的值，这里的字段名大小写不敏感tracking_column => "update_time"#同步点文件，这个文件记录了上次的同步点，重启时会读取这个文件，这个文件可以自定义last_run_metadata_path => "/opt/metadata/product_table"#是否分页查询jdbc_paging_enabled => "true"#每次分页的数量jdbc_page_size => "100"#为多数据表/源做准备，作为区分条件type => "type_expert"}
}
output {if [type] == "type_user" {elasticsearch {# ES的IP地址及端口hosts => ["192.168.114.136:9200"]# 索引名称 可自定义index => "user"# 需要关联的数据库中有有一个id字段，对应类型中的iddocument_id => "%{user_id}"# 文档类型（不建意使用，因为elasticsearch 7.0.0以后将不建议使用，8.0.0 以后完全不支持。我是直接删除了）document_type => "user"}}if [type] == "type_expert" {elasticsearch {# ES的IP地址及端口hosts => ["192.168.114.136:9200"]# 索引名称 可自定义index => "product"# 需要关联的数据库中有有一个id字段，对应类型中的iddocument_id => "%{id}"# 文档类型（不建意使用，因为elasticsearch 7.0.0以后将不建议使用，8.0.0 以后完全不支持。我是直接删除了）document_type => "expert"}}stdout {# 以JSON格式输出至elasticsearchcodec => json_lines}
}

6、启动Logstash

采用yum与RPM方式安装的请往这里看
- 修改logstash的pipelines.yml配置文件
- 进入配置文件后首先删除原有的配置信息
- 然后新增我们自己的配置文件信息
```
- pipeline.id: table1path.config: "/etc/logstash/mysql_config/table_1.conf"
```
- 保存并退出
- 对新建用户授予权限
```
chown -R cuu:es_user /etc/logstash
chown -R cuu:es_user /usr/share/logstash
```
- 刷新服务配置
```
systemctl daemon-reload
```
- 输入启动命令
```
systemctl start logstash.service
```

采用压缩包形式安装的请往这里看

如果只有一份conf文件启动方式为

前台启动

/opt/logstash-7.5.2/bin/logstash -f /opt/logstash-7.5.2/mysql_config/table_1.conf

后台启动

nohup /opt/logstash-7.5.2/bin/logstash -f /opt/logstash-7.5.2/mysql_config/table_1.conf &

如果有多分conf文件，则需要在默认配置文件下加入相关信息，如果只有一份配置文件，也可以使用这种方式（推荐，启动简单，看进程配置不繁琐）

修改logstash的pipelines.yml配置文件

vim  /opt/logstash-7.5.2/config/pipelines.yml

在打开的文件最下方加入，以词类推，这里的代码块最前面有个横杠别忘记写入了。

- pipeline.id: table1path.config: "/opt/logstash-7.5.2/mysql_config/table_1.conf"
- pipeline.id: table2path.config: "/opt/logstash-7.5.2/mysql_config/table_2.conf"
- pipeline.id: table3path.config: "/opt/logstash-7.5.2/mysql_config/table_3.conf"

开始启动logstash
- 前台启动
```
/opt/logstash-7.5.2/bin/logstash
```

后台启动

nohup /opt/logstash-7.5.2/bin/logstash &

7、停止logstash

使用yum与rpm新式安装的可以通过systemctl stop logstash.service命令进行关闭

关闭命令
systemctl stop logstash.service
重启命令
systemctl restart logstash.service
目前，使用压缩包安装的，未找到可以关闭的命令，目前已知使用Kill杀死进程

kill -9 [进程号]

8、设置logstash开机自启

使用yum与rpm新式安装的elasticsearch可以通过systemctl enable logstash.service命令进行设置开机自启

systemctl enable logstash.service

对新用户开放logstash文件位置权限

chown -R cuu:es_user /opt/logstash-7.5.2
刷新服务配置

systemctl daemon-reload
使用压缩包安装的logstash需要在/etc/init.d 创建脚本文件

vim /etc/init.d/logstash

输入代码块，需要修改的地方为logstash的安装位置与用户切换

#!/bin/sh
#chkconfig: 2345 80 05
#description: logstash-7case "$1" in
start)su -  cuu<<!cd /opt/logstash-7.5.2/bin#这里启动方式可以根据实际的配置来设置启动命令# 例如：nohup ./logstash -f /opt/logstash-7.5.2/mysql_config/table_1.conf &nohup ./logstash &
!echo "logstash startup";;  
stop)logstash_pid=`ps aux|grep logstash | grep -v 'grep logstash' | awk '{print $2}'`kill -9 logstash_pidecho "logstash stopped";;  
restart)logstash_pid=`ps aux|grep logstash | grep -v 'grep logstash' | awk '{print $2}'`kill -9 logstash_pidecho "logstash stopped"su - cuu<<!cd /opt/logstash-7.5.2/bin#这里启动方式可以根据实际的配置来设置启动命令# 例如：nohup ./logstash -f /opt/logstash-7.5.2/mysql_config/table_1.conf &nohup ./logstash &
!echo "logstash startup";;  
*)echo "start|stop|restart";;  
esacexit $?

写入完毕后，保存退出。

在 /etc/init.d 目录下赋予新创建的logstash文件执行权限
chmod +x /etc/init.d/logstash
添加到开机启动任务
chkconfig --add logstash
重启验证是否成功启动，如果失败请验证用户是否切换正确，logstash地址是否正确
```
ps -ef|grep logstash
```
- 出现下图表示启动成功，为什么我这里文字多一点的，因为我这里使使用自定义配置文件启动的，在启动命令后面加了配置文件地址。