LogStash如何通过jdbc 从mysql导入elasticsearch

input {
    stdin {
    }
    jdbc {
      # mysql jdbc connection string to our backup databse
      jdbc_connection_string => "jdbc:mysql://localhost:3306/userdb?useUnicode=true&characterEncoding=utf-8&useSSL=false"
      # the user we wish to excute our statement as
      jdbc_user => "user"
      jdbc_password => "pass"
      # the path to our downloaded jdbc driver
      jdbc_driver_library => "mysql-connector-java-5.1.40-bin.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement_filepath => "mysqltestsql.sql"
      schedule => "* * * * *"
      type => "jdbc"
      use_column_value => true
      tracking_column => id
      last_run_metadata_path => "C:\logstash-5.2.0\bin\last_run_metadata_path\last_run_metadata_path"
    }
}

filter {
    json {
        source => "message"
        remove_field => ["message","_id"]
    }
}

output {  
    elasticsearch {
        action => "index"
        hosts => ["http://localhost:9200"] 

        index => "mysqltestdb2"
        #index => "user"
        workers => 1
    }
     #stdout {
         #codec => rubydebug
         #codec => json_lines
     #}
}

select id, name from userinfo WHERE id > :sql_last_value