首先进入到 Logstash/bin 文件夹下，我们的配置文件是放在Logstash/bin/mysql/jdbc-timedtask.conf 文件中。
目录结构：
配置文件信息 Logstash/bin/mysql/jdbc.conf：
input {
jdbc {
#索引的类型,标识符。
type => "user"
# jdbc 驱动包位置
jdbc_driver_library => "./mysql/mysql-connector-java-5.1.49.jar"
# 要使用的驱动包类
jdbc_driver_class => "com.mysql.jdbc.Driver"
#配置jdbc地址，serverTimezone=Asia/Shanghai 时区设置为上海。
jdbc_connection_string => "jdbc:mysql://localhost:3306/db?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai"
# 数据库用户名
jdbc_user => "root"
# 数据库密码
jdbc_password => "123456"
#定时任务 分别代表(分 时 天 月 年)。按需配置。
schedule => "35 10 * * *"
#logstash从mysql同步数据，如果数据量太大时可以进行分页同步。
jdbc_paging_enabled => true
#分页大小
jdbc_page_size => 100000
#流式获取数据，每次取10000
jdbc_fetch_size => 10000
#尝试连接到数据库的最大次数
connection_retry_attempts => 3
#连接尝试之间休眠的秒数
connection_retry_attempts_wait_time => 1
#超时时间
jdbc_pool_timeout => 5
#是否强制字段为小写的形式
lowercase_column_names => true
#-------下面的配置是设置增量同步数据的-----
#启用数据库字段记录增量
use_column_value => true
#数据库同步时按照那个字段为依据进行增量同步。我用的是操作时间字段。
tracking_column => operatetime
#递增字段的类型，numeric 表示数值类型, timestamp 表示时间戳类型
tracking_column_type => "timestamp"
##保存上次运行记录，增量提取数据时使用
record_last_run => true
#记录上次的同步点，需要事先创建好，重启时会读取这个文件，这个文件可以手动修改
last_run_metadata_path => "./mysql/jdbc-user.txt"
#是否清除last_run_metadata_path的记录,如果是true,则从头开始查询所有的数据库记录
clean_run => false
#按照什么语句进行增量同步数据。可以使用sql文件，也可以使用sql语句，我这里使用的是sql文件。
statement_filepath => "./mysql/timetask-user.sql"
# statement => "select * from user where operate_time >=:sql_last_value and operate_time <now()"
#设置时区,
jdbc_default_timezone => "Asia/Shanghai"
}
#多表进行增量同步数据，可以和上面的配置一致。
jdbc {
type => "department"
jdbc_driver_library => "./mysql/mysql-connector-java-5.1.49.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/db?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai"
jdbc_user => "root"
jdbc_password => "123456"
schedule => "35 10 * * *"
jdbc_paging_enabled => true
jdbc_page_size => 100000
jdbc_fetch_size => 10000
connection_retry_attempts => 3
connection_retry_attempts_wait_time => 1
jdbc_pool_timeout => 5
lowercase_column_names => true
#codec => plain {charset => "UTF-8"}
use_column_value => true
tracking_column => id
#这里的是以Id作为增量字段的。类型是bigint，这里使用的是numeric。
tracking_column_type => "numeric"
record_last_run => true
last_run_metadata_path => "./mysql/jdbc-position.txt"
clean_run => false
statement_filepath => "./mysql/timetask-department.sql"
jdbc_default_timezone => "Asia/Shanghai"
}
}
filter {
json {
source => "message"
}
date{
#格式化时间类型
match => ["operatetime","yyy-MM-dd HH:mm:ss"]
}
if[type]=="company"{
#虽然使用了上海的时区，但是logstash同步数据的时候还是会差8个小时.
ruby {
code => "event.set(‘operatetime‘, event.get(‘@timestamp‘).time.localtime + 8*60*60)"
}
ruby {
code => "event.set(‘@timestamp‘,event.get(‘operatetime‘))"
}
mutate {
remove_field => ["timestamp"]
# rename => { "[host][name]" => "host" }
}
}
}
output {
if[type] == "user"{
elasticsearch{
# es host : host，如果是集群，就用“,”分开，例如["127.0.0.1:9200", "127.0.0.2:9200"] ，我这里就一个节点。
hosts => ["http://127.0.0.1:9200"]
# 索引 ，logstash 的索引名称。
index => "user"
# logstash 的主键。
document_id => "%{id}"
document_type => "db"
#elasticsearch 设置密码后，logstash需要登录时的账号密码。
user => elastic
password => 123456gw
}
}
if[type] == "department"{
elasticsearch{
hosts => ["http://127.0.0.1:9200"]
index => "department"
document_id => "%{id}"
document_type => "db"
user => elastic
password => 123456gw
}
}
}
Logstash/bin/mysql/mysql-connector-java-5.1.49.jar
Logstash/bin/mysql/jdbc-user.sql
select * from user where operate_time >=:sql_last_value and operate_time <now()
Logstash/bin/mysql/jdbc-department.sql
select * from department where id >=:sql_last_value
Logstash/bin/mysql/jdbc-user.txt 数据库中操作时间最小的是2019年的，所以设置为2019年。
--- !ruby/object:DateTime ‘2019-01-01 00:00:01.000000000 Z‘
--- 0
原文：https://www.cnblogs.com/Glc-Blog/p/14177222.html