环境
Ubuntu：18.04
以下服务都是安装在docker里
logstash： 7.1.1
mysql: 5.7
elasticsearch：7.1.1
安装logstash-input-jdbc
#查看是否已经安装logstash-input-jdbc,如果有输出说明已经安装了
logstash-plugin list | grep logstash-input-jdbc
#如果没有安装logstash-input-jdbc,用下面命令安装
logstash-plugin install logstash-input-jdbc
下载mysql驱动
wget https://mysql.com//Downloads/Connector-J/mysql-connector-java-5.1.46.zip
#解压
unzip mysql-connector-java-5.1.46.zip
配置logstash
找到pipeline配置文件(配置文件详解:https://segmentfault.com/a/1190000016591126)
编辑
input {
beats {
port => 5044
}
jdbc {
clean_run => false
jdbc_default_timezone => "Asia/Shanghai"
#定时，类似与crontab
schedule => "* * * * *"
type => "goods"
#上面解压出来的mysql驱动文件路径，指向mysql-connector-java-5.1.46-bin.jar这个文件
jdbc_driver_library => "/usr/share/logstash/driver/mysql/mysql-connector-java-5.1.46-bin.jar"
#驱动类名
jdbc_driver_class => "com.mysql.jdbc.Driver"
#数据库连接信息，jdbc:mysql://127.0.0.1:3306/数据库?useUnicode=true&characterEncoding=utf-8&useSSL=false
jdbc_connection_string => "jdbc:mysql://mysql5:3306/wctxg008useUnicode=true&characterEncoding=utf-8&useSSL=false"
#数据库账号
jdbc_user => "root"
#数据库密码
jdbc_password => "root"
#sql查询语句
statement => "SELECT * FROM ds_goods where update_time>:last_sql"
# 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
record_last_run => true
# 这里如果 tracking_column是用时间追踪比如：数据的更新时间或创建时间等和时间有关的这里一定不能是true, 切记切记切记，我是用update_time来追踪的
use_column_value => false
#追踪的字段
tracking_column => "update_time"
#如果use_column_value为true则值为numeric，否则可不设置此值
tracking_column_type => "timestamp"
# 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
# clean_run => false
# 是否将 字段(column) 名称转小写
lowercase_column_names => false
}
}
output {
if [type] == 'goods' {
elasticsearch {
hosts => ["http://elasticsearch:9200"]
index => "goods"
}
}
}
