3-3-关系型数据库和elastic之间操作
1. 数据备份至关系型数据库中个
由于部分重要日志需要按照开发的要求将日志存储到数据库中。这个实验以上面一个环境为基础。加入数据库服务器
1. 1安装mariadb数据库
安装数据库并启动
root@mysql:~# apt install mariadb-server root@mysql:~# vi /etc/mysql/mariadb.conf.d/50-server.cnf bind-address = 0.0.0.0 root@mysql:~# systemctl restart mariadb 创建一个数据库并授权
root@mysql:~# mysql Welcome to the MariaDB monitor. Commands end with ; or \g. Your MariaDB connection id is 31 Server version: 10.6.16-MariaDB-0ubuntu0.22.04.1 Ubuntu 22.04 Copyright (c) 2000, 2018, Oracle, MariaDB Corporation Ab and others. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. MariaDB [(none)]> create database elk character set utf8 collate utf8_bin; Query OK, 1 row affected (0.001 sec) MariaDB [(none)]> grant all privileges on elk.* to elk@"192.168.71.%" identified by '123456'; Query OK, 0 rows affected (0.004 sec)
1.2. 在logstash主机上安装jdbc插件
安装
logstash-output-jdbc插件root@logstash03:~# /usr/share/logstash/bin/logstash-plugin list | grep jdbc logstash-integration-jdbc ├── logstash-input-jdbc ├── logstash-filter-jdbc_streaming └── logstash-filter-jdbc_static root@logstash03:~# logstash-plugin install logstash-output-jdbc logstash-plugin: command not found root@logstash03:~# /usr/share/logstash/bin/logstash-plugin install logstash-output-jdbc Using bundled JDK: /usr/share/logstash/jdk Validating logstash-output-jdbc Resolving mixin dependencies Installing logstash-output-jdbc Installation successful在官网下载
jar包MySQL Connector/J是MySQL官方JDBC驱动程序JDBC(Java Data Base Connectivity,java数据库连接)是一种用于执行SQL语句的Java API, 可以为多种关系数据库提供统一访问 , 它由一组用Java语言编写的类和接口组成。官方下载地址 : https://dev.mysql.com/downloads/
下载对应系统版本的包 :
mysql-connector-java-8.0.19-1.el7.noarch
创建一个存放
jar包的目录 , 将下载的包进行安装 , 并将生成的jar复制到指定创建的目录下root@logstash04:~# dpkg -c mysql-connector-j_8.3.0-1ubuntu22.04_all.deb drwxr-xr-x root/root 0 2023-12-13 03:09 ./ drwxr-xr-x root/root 0 2023-12-13 03:09 ./usr/ drwxr-xr-x root/root 0 2023-12-13 03:09 ./usr/share/ drwxr-xr-x root/root 0 2023-12-13 03:09 ./usr/share/doc/ drwxr-xr-x root/root 0 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/ -rw-r--r-- root/root 93960 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/CHANGES.gz -rw-r--r-- root/root 174 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/INFO_BIN -rw-r--r-- root/root 134 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/INFO_SRC -rw-r--r-- root/root 22625 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/LICENSE.gz -rw-r--r-- root/root 1624 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/README -rw-r--r-- root/root 210 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/changelog.Debian.gz -rw-r--r-- root/root 356 2023-12-13 03:09 ./usr/share/doc/mysql-connector-j/copyright drwxr-xr-x root/root 0 2023-12-13 03:09 ./usr/share/java/ -rw-r--r-- root/root 2495515 2023-12-13 03:09 ./usr/share/java/mysql-connector-j-8.3.0.jar lrwxrwxrwx root/root 0 2023-12-13 03:09 ./usr/share/java/mysql-connector-java-8.3.0.jar -> mysql-connector-j-8.3.0.jar root@logstash04:~# dpkg -i mysql-connector-j_8.3.0-1ubuntu22.04_all.deb Selecting previously unselected package mysql-connector-j. (Reading database ... 124831 files and directories currently installed.) Preparing to unpack mysql-connector-j_8.3.0-1ubuntu22.04_all.deb ... Unpacking mysql-connector-j (8.3.0-1ubuntu22.04) ... Setting up mysql-connector-j (8.3.0-1ubuntu22.04) ... root@logstash03:~# dpkg -i mysql-connector-j_8.3.0-1ubuntu22.04_all.deb Selecting previously unselected package mysql-connector-j. (Reading database ... 124831 files and directories currently installed.) Preparing to unpack mysql-connector-j_8.3.0-1ubuntu22.04_all.deb ... Unpacking mysql-connector-j (8.3.0-1ubuntu22.04) ... Setting up mysql-connector-j (8.3.0-1ubuntu22.04) ... root@logstash04:~# ll /usr/share/java/mysql-connector-j-8.3.0.jar -rw-r--r-- 1 root root 2495515 Dec 13 03:09 /usr/share/java/mysql-connector-j-8.3.0.jar root@logstash04:~# cp /usr/share/java/mysql-connector-j-8.3.0.jar /usr/share/logstash/vendor/jar/jdbc/ root@logstash04:~# ll /usr/share/logstash/vendor/jar/jdbc/ total 2448 drwxr-xr-x 2 root root 4096 Apr 14 09:56 ./ drwxr-xr-x 3 root root 4096 Apr 14 09:53 ../ -rw-r--r-- 1 root root 2495515 Apr 14 09:56 mysql-connector-j-8.3.0.jar root@logstash03:~# mkdir /usr/share/logstash/vendor/jar/jdbc -p root@logstash03:~# cp /usr/share/java/mysql-connector-j-8.3.0.jar /usr/share/logstash/vendor/jar/jdbc/ 重启
logstash服务 , 要保证此时的logstash服务要能够启动root@logstash04:~# systemctl restart logstash
1.3. 在mariadb数据库创建收集日志表
root@mysql:~# mysql
Welcome to the MariaDB monitor. Commands end with ; or \g.
Your MariaDB connection id is 32
Server version: 10.6.16-MariaDB-0ubuntu0.22.04.1 Ubuntu 22.04
Copyright (c) 2000, 2018, Oracle, MariaDB Corporation Ab and others.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
MariaDB [(none)]> use elk
Database changed
MariaDB [elk]> create table elklog (clientip varchar(64),responsetime varchar(16),http_host varchar(64),http_user_agent varchar(256),url varchar(128),time timestamp(0));
Query OK, 0 rows affected (0.041 sec)
# # 创建收集日志的elklog表 , 需要定义字段名称1.4. 配置logstash
这个步骤主要是希望, 输出部分的内容同时可以保存在mysql服务器中
在logstash服务器的/etc/logstash/conf.d目录下创建一个收集nginx日志的配置文件。主要在logstash03、logstash04节点上面配置
root@logstash03:~# cat /etc/logstash/conf.d/nginx-to-redis.conf
input {
beats {
port => 5044
#if [fields][app]=="nginx"{
#codec => json
#}
}
}
output {
#stdout{}
if [fields][app]=="syslog" {
redis {
host => "192.168.71.43"
port => "6379"
password => "123456"
db => "1"
key => "linux-syslog-key"
data_type => "list"
}
#stdout{}
}
if [fields][app]=="nginx" {
redis {
host => "192.168.71.43"
port => "6379"
password => "123456"
db => "2"
key => "linux-nginx-key"
data_type => "list"
}
#stdout {}
jdbc {
connection_string => "jdbc:mysql://192.168.71.45/elk?user=elk&password=123456&useUnicode=true&characterEncoding=UTF8"
statement => ["INSERT INTO elklog(clientip,responsetime,http_host,http_user_agent,url) VALUES(?,?,?,?,?)", "clientip","responsetime","http_host","http_user_agent","url"]
}
}
}
1.5. 在数据库中查看收集到的nginx日志
MariaDB [elk]> select * from elklog;
+---------------+--------------+---------------+-----------------+-------------+---------------------+
| clientip | responsetime | http_host | http_user_agent | url | time |
+---------------+--------------+---------------+-----------------+-------------+---------------------+
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:22 |
| 192.168.71.37 | 0 | 192.168.71.44 | NULL | /index.html | 2024-04-14 15:02:32 |
.......
2. 关系型数据库数据提取到elasticsearch中
Logstash Java数据库连接 (JDBC) 输入插件使你能够从许多流行的关系数据库(包括 MySQL和 Postgres)中提取数据。 从概念上讲 , JDBC输入插件运行一个循环 , 该循环定期轮询关系数据库以查找自该循环的最后一次迭代以来插入或修改的记录。
这里实验的环境沿用上一步的实验过程。这里包括准备mysql环境和下载对应的jdbc插件
2.1 配置数据库信息
对于这个例子 , 让我们创建一个新的数据库es_db和表es_table , 作为我们的Elasticsearch数据的来源。
运行以下
SQL语句以生成具有三个列表的新MySQL数据库:CREATE DATABASE es_database; USE es_database; DROP TABLE IF EXISTS es_table; CREATE TABLE es_table ( id BIGINT(20) UNSIGNED NOT NULL, PRIMARY KEY (id), UNIQUE KEY unique_id (id), client_name VARCHAR(32) NOT NULL, modification_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); MariaDB [(none)]> grant all privileges on es_database.* to elk@"192.168.71.%" identified by '123456'; Query OK, 0 rows affected (0.004 sec)
考虑如何处理删除以及如何将它们通知
Elasticsearch。 通常 , 删除一条记录会导致它立即从MySQL数据库中删除。 没有删除的记录。Logstash未检测到更改 , 因此该记录保留在Elasticsearch中有两种可能的方法来解决这个问题:
你可以在源数据库中使用 “软删除(
soft deletes)”。 本质上 , 记录首先通过布尔标志标记为删除。 当前正在使用你的源数据库的其他程序必须在其查询中过滤掉 “软删除”。 “软删除” 被发送到Elasticsearch, 在那里可以处理它们。 之后 , 你的源数据库和Elasticsearch都必须删除这些 “软删除”。你可以定期清除基于数据库的
Elasticsearch索引 , 然后使用新摄取的数据库内容刷新Elasticsearch。
登录到你的
MySQL服务器并将三个记录添加到你的新数据库:use es_database INSERT INTO es_table (id, client_name) VALUES (1,"Targaryen"), (2,"Lannister"), (3,"Stark");
使用
SQL语句验证你的数据:select * from es_table;
现在 , 让我们回到Logstash并将其配置为接收此数据。
2.2 使用 JDBC 输入插件配置 Logstash 管道
让我们设置一个示例Logstash输入管道 , 以从新的JDBC插件和MySQL数据库中获取数据。 除了MySQL , 你还可以从任何支持 JDBC的数据库输入数据。
在
<localpath>/logstash-8.5.0/中 , 创建一个名为jdbc.conf的新文本文件。将以下代码复制并粘贴到这个新的文本文件中。 此代码通
JDBC插件创建一个Logstash管道。input { jdbc { jdbc_driver_library => "<driverpath>/mysql-connector-java-<versionNumber>.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://<MySQL host>:3306/es_db" jdbc_user => "<myusername>" jdbc_password => "<mypassword>" jdbc_paging_enabled => true tracking_column => "unix_ts_in_secs" use_column_value => true tracking_column_type => "numeric" schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "@version", "unix_ts_in_secs"] } } output { stdout { codec => "rubydebug"} }说明:
指定本地
JDBC驱动程序.jar文件的完整路径(包括版本号)。 例如:jdbc_driver_library => "/usr/share/java/mysql-connector-j-8.3.0.jar"提供
MySQL主机的IP地址或主机名和端口。 例如 ,jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/es_database"提供您的
MySQL凭据。 用户名和密码都必须用引号引起来。
注意: 如果你使用的是
MariaDB(MySQL的一个流行的开源社区分支) , 你需要做一些不同的事情:代替
MySQL JDBC驱动程序 , 下载并解压缩MariaDB的JDBC驱动程序。替换
jdbc.conf代码中的以下行 , 包括最后一行中的ANSI_QUOTES片段:jdbc_driver_library => "<driverPath>/mariadb-java-client-<versionNumber>.jar"jdbc_driver_class => "org.mariadb.jdbc.Driver"jdbc_connection_string => "jdbc:mariadb://<mySQLHost>:3306/es_db?sessionVariables=sql_mode=ANSI_QUOTES
以下是有关
Logstash管道代码的一些其他详细信息:我们需要把下载的 JDBC 驱动拷贝到如下的目录中:
root@logstash03:~# cp /usr/share/java/mysql-connector-j-8.3.0.jar /usr/share/logstash/logstash-core/lib/jars/ root@logstash03:~# ls /usr/share/logstash/logstash-core/lib/jars/mysql-connector-j-8.3.0.jar /usr/share/logstash/logstash-core/lib/jars/mysql-connector-j-8.3.0.jar根据我的情况 , 我的
jdbc.conf的代码如下:input { jdbc { jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://192.168.71.45:3306/es_database" jdbc_user => "elk" jdbc_password => "123456" jdbc_paging_enabled => true tracking_column => "unix_ts_in_secs" use_column_value => true tracking_column_type => "numeric" schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "@version", "unix_ts_in_secs"] } } output { stdout { codec => "rubydebug"} }使用新的
JDBC配置文件启动Logstashroot@logstash03:~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/jdbc.confLogstash通过标准输出 (stdout) , 你的命令行界面输出您的MySQL数据。 初始数据加载的结果应类似于以下内容:
Logstash结果会定期显示SQL SELECT语句 , 即使在MySQL数据库中没有任何新内容或修改内容时也是如此:
打开你的
MySQL控制台。 让我们使用以下SQL语句将另一条记录插入该数据库:use es_database INSERT INTO es_table (id, client_name) VALUES (4,"Baratheon");
切换回你的
Logstash控制台。Logstash检测到新记录 , 控制台显示类似如下结果：
查看
Logstash输出结果以确保你的数据看起来正确。 使用CTRL+C关闭Logstash。
2.3 输出到elasticsearch
在本节中 , 我们配置Logstash以将MySQL数据发送到Elasticsearch。 我们修改在使用JDBC输入插件配置 Logstash管道部分中创建的配置文件 , 以便数据直接输出到Elasticsearch。 我们启动Logstash发送数据 , 然后登录Kibana来验证数据。
打开
Logstash文件夹下的jdbc.conf文件进行编辑。用以下内容更新输出部分input { jdbc { jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://localhost:3306/es_database" jdbc_user => "root" jdbc_password => "1234" jdbc_paging_enabled => true tracking_column => "unix_ts_in_secs" use_column_value => true tracking_column_type => "numeric" schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "@version", "unix_ts_in_secs"] } } output { stdout { codec => "rubydebug"} elasticsearch { hosts => ["http://192.168.71.29:9200"] index => "data-%{+YYYY.MM.dd}" user => "elastic" password => "123456" } }现在 , 如果您只是像使用新输出一样重新启动
Logstash, 则不会有MySQL数据发送到我们的Elasticsearch索引。Logstash保留了以前的sql_last_value时间戳 , 并看到从那时起MySQL数据库中没有发生新的更改。 因此 , 根据我们配置的SQL查询 , 没有新数据要发送到Logstash。解决方案：在
jdbc.conf文件的 JDBC 输入部分添加clean_run => true作为新行。 当设置为true时 , 此参数会将sql_last_value重置回零。input { jdbc { ... clean_run => true ... } }在将
clean_run设置为true运行一次Logstash后 , 你可以删除clean_run行 , 除非您希望在每次重新启动Logstash时再次发生重置行为。打开命令行界面实例 , 转到您的
Logstash安装路径 , 然后启动Logstash：root@logstash03:~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/jdbc.conf运行上面的命令后 ,
Logstash就会把数据传入到Elasticsearch中。
2.4 在 Kibana 中检查数据
我们在Kibana的console中打入如下的命令:
GET data-*/_search
我们看到有四个数据已经被写入到·Elasticsearch·中了。
参考链接