Logstash-input-jdbc 同步 mysql 准实时数据至 ElasticSearch 搜索引擎

logstash-input-jdbc 插件将 Zabbix 数据库中 alerts 表告警数据推送至 ElasticSearch 搜索引擎。

Zabbix 数据库

alerts 表结构

alerts 表数据条目数

logstash-input-jdbc 插件

安装 logstash-input-jdbc 插件

代码语言：javascript复制

/usr/share/logstash/bin/logstash-plugin  install logstash-input-jdbc

查看 logstash 插件列表

代码语言：javascript复制

/usr/share/logstash/bin/logstash-plugin  list

升级 logstash-input-jdbc 插件

代码语言：javascript复制

/usr/share/logstash/bin/logstash-plugin  update  logstash-input-jdbc

mysql-connector-java 驱动程序

mysql-connector-java 下载地址

https://dev.mysql.com/downloads/connector/j/

安装 mysql-connector-java 驱动

代码语言：javascript复制

rpm -ivh mysql-connector-java-8.0.21-1.el7.noarch.rpm

拷贝 mysql-connector-java.jar

代码语言：javascript复制

cp  /usr/share/java/mysql-connector-java.jar /usr/share/logstash/logstash-core/lib/jars/

logstash 配置

代码语言：javascript复制

cat  /etc/logstash/conf.d/odbc.conf 
input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://192.168.99.51:3306/zabbix"
    jdbc_default_timezone => "Asia/Shanghai"
    jdbc_user => "zabbix"
    jdbc_password => "******"
    schedule => "* * * * *"
    jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    use_column_value => false
    tracking_column => "clock"
    statement => "select * from alerts  where from_unixtime(clock) > :sql_last_value"
    tracking_column_type => "timestamp"
    type => "zabbix_alters"
  }
}

filter {
  json {
    source => "message"
  }
}

output {
  stdout {
    codec => json
  }
if [type] == 'zabbix_alters' {
  elasticsearch { 
    index => "zabbix_alters-%{ YYYY.MM.dd}"
    hosts => ["192.168.99.185:9200"]
    document_id => "%{alertid}"
    user => "elastic"
    password => "*******"    
   }
 }
}

odbc.conf 参数介绍

jdbc_connection_string ： mysql 数据库地址

jdbc_default_timezone ： 设定时区

jdbc_user ：数据库用户名

jdbc_password ：数据库密码

schedule ：执行SQL任务，类似crontab

jdbc_driver_library ：jdbc mysql-connector-java 驱动路径

jdbc_driver_class ：驱动类名称

jdbc_paging_enabled => "true" ：数据分页

jdbc_page_size => "50000" ： 数据条目

use_column_value：是否使用字段的值

tracking_column ：监控字段用于区分增加数据

statement ：SQL执行查询语句

tracking_column_type ：区分的字段类型

type ：类型 判断output es 写入标识

运行 logstash odbc.conf

代码语言：javascript复制

/usr/share/logstash/bin/logstash -f odbc.conf

Kibana

Zabbix 告警数据写入数据库 alerts 表后 ，logstash-input-jdbc 插件通过执行 SQL 语句将数据同步到 elasticsearch ，同时使用递增字段做条件查询，记录字段当前的查询位置进行数据增量同步，由于 SQL 任务1分钟执行一次到 elasticsearch 存储 kibana 展示的 数据存在1分钟左右的延迟。

JSON 数据

查看索引数据

代码语言：javascript复制

GET /zabbix_alters-2020.08.21/_search
{
  "query": {
    "match_all": {}
  }
}

logstash-input-jdbc 插件官方文档

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html

jdbc java 数据库 sql 云数据库SQLServer

0 人点赞