大数据-186 Elasticsearch - ELK 家属 Logstash Input插件 JDBC & syslog ...

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
  
• HDFS（已更完）
  
• MapReduce（已更完）
  
• Hive（已更完）
  
• Flume（已更完）
  
• Sqoop（已更完）
  
• Zookeeper（已更完）
  
• HBase（已更完）
  
• Redis （已更完）
  
• Kafka（已更完）
  
• Spark（已更完）
  
• Flink（已更完）
  
• ClickHouse（已更完）
  
• Kudu（已更完）
  
• Druid（已更完）
  
• Kylin（已更完）
  
• Elasticsearch（正在更新…）
  

章节内容

上节我们完成了如下的内容：

• Logstash 安装设置
  
• Logstash Input 插件
  

根本介绍

Logstash 的 JDBC 和 Syslog 是两种不同的 Input 插件，它们分别用于从数据库和 syslog 日志中收集数据。在详细介绍之前，首先明确一下它们各自的用途和特点：

• JDBC 插件：用于从数据库（如 MySQL、PostgreSQL、SQL Server 等）中提取数据。它使用 JDBC 驱动毗连数据库并运行 SQL 查询，收集查询结果并将其输入到 Logstash 举行处置惩罚和发送。
  
• Syslog 插件：用于从 Syslog（系统日志）服务中收集日志数据。Syslog 是一种标准化的日志传输协议，广泛用于网络设备、利用系统和应用的日志记载。
  

JDBC Input 插件

JDBC 插件用于从关系型数据库中提取数据，特别适用于将结构化的业务数据导入到 ELK（Elasticsearch, Logstash, Kibana）堆栈中。它支持定时调度和增量提取，适合数据同步和 ETL 场景。
主要功能：

• 数据库毗连：使用 JDBC 驱动来毗连各种数据库（例如 MySQL、PostgreSQL、Oracle、SQL Server 等）。
  
• SQL 查询：答应用户通过 SQL 查询语句选择数据，支持复杂的查询条件。
  
• 增量提取：可以设置“追踪列”（tracking column），基于某一列（如自增 ID 或时间戳）实现增量数据拉取，避免重复导入。
  
• 定时调度：可以设置定时任务，定期查询和同步数据库的数据。
  
• 错误重试机制：可以处置惩罚毗连错误并主动重试，确保数据采集的稳定性。
  

Syslog Input 插件

Syslog 插件用于吸收基于 Syslog 协议传输的日志数据。Syslog 是一种广泛使用的日志协议，特别是在网络设备和 Linux/Unix 利用系统中。Logstash 的 Syslog 插件能够监听特定的端口，吸收和剖析这些日志。
主要功能：

• 协议支持：支持 Syslog 协议，包罗 RFC 3164 和 RFC 5424 格式的日志。
  
• 多种传输方式：支持通过 TCP 和 UDP 协议吸收日志，适应不同的传输需求。
  
• 日志剖析：主动剖析 Syslog 消息的头部字段，例如时间戳、主机名、步伐名等。
  
• 网络监听：可以通过监听指定的 IP 和端口，持续吸收并处置惩罚来自网络的 Syslog 日志。
  

JDBC插件

JDBC插件可以采集某张数据库表当中的数据到Logstash当中来
准备数据

1. CREATE TABLE `user` (
2.   `id` bigint(20) NOT NULL,
3.   `user_name` varchar(25) DEFAULT NULL,
4.   `gender` varchar(20) DEFAULT NULL,
5.   `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
6.   PRIMARY KEY(`id`)
7. ) ENGINE=InnoDB DEFAULT CHRASET=utf8;
9. INSERT INTO `user` VALUES(1, 'zhangsan', 'male', '2024-08-16 00:00:00');
10. INSERT INTO `user` VALUES(2, 'lisi', 'female', '2024-08-16 00:00:00');

复制代码

目前我们的MySQL服务器是在 h122 节点上的（之前给Hive和HBase等业务使用的），现在我们必要到 h122 数据库中，执行上述的SQL指令。
我这里使用Navicat执行， 执行结果如下图所示：

编写设置

1. cd /opt/servers/logstash-7.3.0/config
2. vim jdbc.conf

复制代码

写入如下的内容（这里留意些自己的环境，我的环境和你的不一样）：

1. input {
2.   jdbc {
3.     jdbc_driver_library => "/opt/servers/mysql-connector-java-8.0.19.jar"
4.     jdbc_driver_class => "com.mysql.jdbc.Driver"
5.     jdbc_connection_string => "jdbc:mysql://h122.wzk.icu:3306/es-test"
6.     jdbc_user => "hive"
7.     jdbc_password => "hive@wzk.icu"
8.     use_column_value => "true"
9.     clean_run => "false"
10.     record_last_run =>"true"
11.     tracking_column => "id"
12.     schedule => "* * * * *"
13.     last_run_metadata_path => "/opt/servers/es/.Logstash_user_jdbc_last_run"
14.     statement => "SELECT * from user where id > :sql_last_value;"
15.   }
16. }
18. output{
19.   stdout{
20.     codec=>rubydebug
21.   }
22. }

复制代码

查抄设置

1. cd /opt/servers/logstash-7.3.0
2. bin/logstash -f /opt/servers/logstash-7.3.0/config/jdbc.conf -t

复制代码

执行结果如下图所示：

启动服务

1. cd /opt/servers/logstash-7.3.0
2. bin/logstash -f /opt/servers/logstash-7.3.0/config/jdbc.conf

复制代码

启动结果如下图所示：

可以看到获取到了对应的数据：

发送数据

现在向数据库中写入数据，就可以发现Logstash监听到了：

1. INSERT INTO `user` VALUES(3, 'wangwu', 'female', '2024-08-16 00:00:00');

复制代码

对应的Logstash的变化：

syslog插件

syslog机制负责记载内核和应用步伐产生的日志信息，管理员可以通过查看日志记载，来掌握系统状态，默认系统已经安装了rsyslog，直接启动即可。
编写设置

创建新脚本，syslog.conf

1. cd /opt/servers/logstash-7.3.0/config
2. vim syslog.conf

复制代码

写入如下的内容：

1. input {
2.   tcp {
3.     port => 6789
4.     type => "syslog"
5.   }
6.   udp {
7.     port => 6789
8.     type => "syslog"
9.   }
10. }
12. filter {
13.   if [type] == "syslog" {
14.     grok {
15.       match => {
16.         "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}"
17.       }
18.       add_field => {
19.         "received_at" => "%{@timestamp}"
20.         "received_from" => "%{host}"
21.       }
22.     }
23.     date {
24.       match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
25.     }
26.   }
27. }
29. output {
30.   stdout {
31.     codec => rubydebug
32.   }
33. }

复制代码

写入的内容如下图所示：

查抄设置

1. cd /opt/servers/logstash-7.3.0
2. bin/logstash -f /opt/servers/logstash-7.3.0/config/syslog.conf -t

复制代码

执行结果如下图所示：

启动服务

1. cd /opt/servers/logstash-7.3.0
2. bin/logstash -f /opt/servers/logstash-7.3.0/config/syslog.conf

复制代码

执行结果如下图所示：

发送数据

修改系统日志设置文件

1. vim /etc/rsyslog.conf

复制代码

添加一行设置：

1. *.* @@h121.wzk.icu:6789

复制代码

写入的效果如下图所示：

重启系统日志服务

1. systemctl restart rsyslog

复制代码

查看数据，可以看到如下的效果：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。