input{
stdin {}
jdbc {
type => "jdbc"
jdbc_connection_string => "jdbc:mysql://192.168.180.63:3306/elasticsearch_view?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=utf8&useSSL=false&allowMultiQueries=true&useTimezone=true&serverTimezone=UTC"
jdbc_user => "root"
jdbc_password => "ABCabc123"
jdbc_driver_library => "E:\\ELK7.12.1\\logstash-7.12.1\\mysql-connector-java-8.0.16.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
# 数据库重连尝试次数
connection_retry_attempts => "3"
# 判断数据库连接是否可用，默认false不开启
jdbc_validate_connection => "true"
# 数据库连接可用校验超时时间，默认3600S
jdbc_validation_timeout => "3600"
# 开启分页查询（默认false不开启）；
jdbc_paging_enabled => "true"
# 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
jdbc_page_size => "2"
tracking_column => tid_time
record_last_run => true
# 差8小时:sql_last_value date_sub(tid_time, interval 8 hour)时间向前8小时date_add(tid_time, interval 8 hour)时间向后8小时
# logstash默认的时间是UTC时间，MySQL中的时区北京时区，所以会相差8个小时
statement => "SELECT id,tid_time,shop_plant, shop_id,shop_name,order_totalfee,product_num,tidcount,fugoulv FROM `a_ck_fugou` WHERE tid_time>= :sql_last_value"
# 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false 默认false；
clean_run => false
# 定时任务，多久执行一次查询，默认一分钟，如果想要没有延迟，分、时、天、月、年，可以使用 schedule => "* * * * *"秒级同步：2秒同步一次schedule => " */2 * * * * "分钟级同步：2分钟同步一次schedule => " 0/2 * * * * "小时级同步：晚上22:00时同步一次schedule => " * 22 * * * "
schedule => "*/1 * * * *"
}
}
filter {}
output {
if [type] == "jdbc" {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "qizidou"
# document_id => "%{KeyId}"
}
}
stdout {
codec => json_lines
}
}
input{
stdin {}
jdbc{
type => "jdbc"
# mysql 数据库的连接信息
jdbc_connection_string => "jdbc:mysql://192.168.180.63:3306/elasticsearch_view?characterEncoding=UTF-8&autoReconnect=true&useTimezone=true&serverTimezone=UTC"
# mysql 用户
jdbc_user => "root"
# 密码
jdbc_password => "ABCabc123"
# jdbc 驱动包位置
jdbc_driver_library => "E:\\ELK7.12.1\\logstash-7.12.1\\mysql-connector-java-8.0.16.jar"
# 要使用的驱动包类
jdbc_driver_class => "com.mysql.jdbc.Driver"
# 处理中文乱码问题
# codec => plain { charset => "UTF-8"}
# 数据库重连尝试次数
connection_retry_attempts => "3"
# 判断数据库连接是否可用，默认false不开启
jdbc_validate_connection => "true"
# 数据库连接可用校验超时时间，默认3600S
jdbc_validation_timeout => "3600"
# 开启分页查询（默认false不开启）；
jdbc_paging_enabled => "true"
# 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
jdbc_page_size => "2"
# 定时任务，多久执行一次查询，默认一分钟，如果想要没有延迟，分、时、天、月、年可以使用 schedule => "* * * * *"
# 定时任务，多久执行一次查询，默认一分钟，如果想要没有延迟，分、时、天、月、年，可以使用 schedule => "* * * * *"秒级同步：2秒同步一次schedule => " */2 * * * * "分钟级同步：2分钟同步一次schedule => " 0/2 * * * * "小时级同步：晚上22:00时同步一次schedule => " * 22 * * * "
schedule => "*/1 * * * *"
# statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
# 执行的sql 文件路径+名称
# statement_filepath => "/opt/data/jdbc.sql"
# sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
# statement_filepath => "mysql/jdbc.sql"
#date_sub(tid_time, interval 8 hour)时间向前8小时date_add(tid_time, interval 8 hour)时间向后8小时 差8个小时statement => "SELECT KeyId,TradeTime,OrderUserName,ModifyTime FROM `DetailTab` WHERE ModifyTime>= :sql_last_value and ModifyTime<now() order by ModifyTime asc"
# logstash默认的时间是UTC时间，MySQL中的时区北京时区，所以会相差8个小时
statement => "SELECT id,tid_time,shop_plant, shop_id,shop_name,order_totalfee,product_num,tidcount,fugoulv FROM `a_ck_fugou` WHERE tid_time>= :sql_last_value"
# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
# lowercase_column_names => false
# Value can be any of: fatal,error,warn,info,debug，默认info；
# sql_log_level => warn
# 追踪的字段（使用MySQL中的时间字段而不使用自增的id字段；因为自增的id字段无法实现更新问题，因为后面需要更新的id值会小于last_run_metadata_path记录的值而无法实现更新操作）
# tracking_column => id
# tracking_column => event_date
# record_last_run => true
# 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
# record_last_run => true
# 需要记录的字段，用于增量同步，需是数据库字段
# tracking_column => "ModifyTime"
# Value can be any of: numeric,timestamp，Default value is "numeric"
# tracking_column_type => timestamp
# record_last_run上次数据存放位置；
# last_run_metadata_path => "mysql/last_id.txt"
# 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
clean_run => false
}
}
output {
if [type] == "jdbc" {
elasticsearch {
# host => "192.168.1.1"
# port => "9200"
# 配置ES集群地址
# hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
# es host : port
hosts => ["127.0.0.1:9200"]
# 索引名字，必须小写
index => "qizidou"
# 数据唯一索引（建议使用数据库KeyID）
# document_id => "%{KeyId}"
}
}
# stdout {
# codec => json_lines
# }
}
|
