Logstash Configuration with JDBC

input {
  jdbc {
    jdbc_driver_library => "/path-to-jdbc/mysql-connector-java-8.0.26/mysql-connector-java-8.0.26.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://ip:port/db"
    jdbc_user => 'user'
    jdbc_password => 'password'
    jdbc_paging_enabled => true
    schedule =>"* * * * *"
    statement => "SELECT cdm_uq.id, cdm_uq.facility_id, cdm_uq.code, cdm_uq.description, cdm_uq.price, cdm_uq.code_original, hospitals.facility_name, hospitals.address, hospitals.city, hospitals.state, hospitals.zip_code, hospitals.phone_number, hospitals.hospital_type, hospitals.emergency_services, hospitals.hospital_overall_rating, hospitals.latitude, hospitals.longitude FROM `cdm_uq` JOIN `hospitals` ON cdm_uq.facility_id = hospitals.facility_id WHERE cdm_uq.id > :sql_last_value"
    use_column_value => true
    tracking_column => "id"
    tracking_column_type => "numeric"
    clean_run => true
    last_run_metadata_path => "/tmp/.logstash_jdbc_last_run"
  }
}
filter {
	if [latitude] and [longitude] {
		mutate { convert => {"latitude" => "float"} }
		mutate { convert => {"longitude" => "float"} }
		mutate { rename => {"latitude" => "[location][lat]"} }
		mutate { rename => {"longitude" => "[location][lon]"} }
	}
}
output {
 stdout { codec =>  "rubydebug"}
  elasticsearch {
      index => "index-name"
      document_id => "%{id}"
  }
}