logstash 整合 es

安装logstash

手动导入Mysql驱动

新建jdbc.conf

vim jdbc.conf

input{
stdin{
}
jdbc{
jdbc_connection_string => "jdbc:mysql://192.168.1.105:3306/logstash_data?characterEncoding=UTF-8&useSSL=false&autoReconnect=true&allowPublicKeyRetrieval=true"
jdbc_user => "root"
jdbc_password => "root"
jdbc_driver_library => "/opt/logstash-7.6.2/config/mysql-connector-java-8.0.11.jar"
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
codec => plain { charset => "UTF-8"}
use_column_value => true
tracking_column => update_date
tracking_column_type => "timestamp"
record_last_run => true
last_run_metadata_path => "./logstash_jdbc_last_run"
jdbc_default_timezone => "Asia/Shanghai"
statement => SELECT * FROM logstash WHERE update_date >= :sql_last_value
clean_run => false
lowercase_column_names => false #是否将字段名称转小写
schedule => "* * * * *"
type => "std"
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {
elasticsearch {
hosts => ["127.0.0.1:9200","192.168.209.161:9200"]
index => "product_index"
document_id => "%{id}"
template_overwrite => true
}
stdout {
codec => json_lines
}
}

启动 logstash

./bin/logstash ./config/jdbc.conf

解决logstash 时间早8小时问题：

ruby {
code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
}
ruby {
code => "event.set('@timestamp',event.get('timestamp'))"
}

#有几个时间就加几个ruby过滤（加在filter{}内）
mutate {
remove_field => ["timestamp"]
}

------------------------------------------------------

logstash 迁移es

logstash主目录下：vim jdbc.conf

input {
elasticsearch {
hosts => ["10.128.120.171", "10.128.120.172", "10.128.120.173", "10.128.120.179", "10.128.120.235"]
# user => "*******"
# password => "*********"
index => "*"
size => 1000
scroll => "1m"
}
}
filter {
mutate {
remove_field => ["@timestamp", "@version"]
}
}
output {
elasticsearch {
hosts => ["10.13.133.121", "10.13.133.122", "10.13.133.123", "10.13.133.124", "10.13.133.125"]
# user => "********"
# password => "**********"
index => "%{[@metadata][_index]}"
}
}