前提需要安装jdk1.8以上
上传logstash-6.4.3.tar.gz到/usr/local目录中
解压

tar –zxvf  logstash-6.4.3.tar.gz

进入logstash

cd logstash-6.4.3 

安装插件1(过程需要一些时间)

bin/logstash-plugin install logstash-input-jdbc

安装插件2(过程需要一些时间)

bin/logstash-plugin install logstash-output-elasticsearch

创建mysql.conf文件

input {
  jdbc {
    jdbc_driver_library => "/usr/local/sql/mysql-connector-java-5.1.46.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://192.168.153.1:3306/test"
    jdbc_user => "root"
    jdbc_password => "root"
    schedule => "* * * * *"
    statement => "SELECT * FROM user WHERE update_time >= :sql_last_value"
    use_column_value => true
    tracking_column_type => "timestamp"
    tracking_column => "update_time"
    last_run_metadata_path => "syncpoint_table"
  }
}


output {
    elasticsearch {
        # ES的IP地址及端口,集群在这里添加地址
        hosts => ["192.168.153.134:9200"]
        # 索引名称 可自定义
        index => "user"
        # 需要关联的数据库中有有一个id字段，对应类型中的id
        document_id => "%{id}"
        document_type => "user"
    }
    stdout {
        # JSON格式输出
        codec => json_lines
    }
}

上传mysql.conf文件到logstash-6.4.3 目录中
上次mysql驱动包到/usr/local/sql目录中

启动logstash

./bin/logstash -f mysql.conf