input {
  jdbc {
    jdbc_driver_library => "/usr/share/java/mysql-connector-java-5.1.28.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/cts_ha1?zeroDateTimeBehavior=convertToNull&ha1?zeroDateTimeBehavior=convertToNull"
    jdbc_user => "root"
    jdbc_password => "root"
    schedule => "* * * * * "
    statement => "select * from `sug_ha1`.`cts_event_control` AS e
JOIN
`sug_ha1`.`cts_event_control_cstm` AS c ON c.id_c = e.id
LEFT JOIN
`cts_ha1`.`events` ctsE ON c.event_id_c = ctsE.event_id
LEFT JOIN
`cts_ha1`.`series` ctsS ON c.series_id_c = ctsS.series_id"
    type => "series_events"
  }
  # add more jdbc inputs to suit your needs
}
filter {
  if["evtlist_date_c"]=="0000-00-00" {
  mutate {
        replace => { "evtlist_date_c" => null }
  }
}
}
output {
                elasticsearch {
                        document_id => "%{id}"
                        index => "ha1"
                        document_type => "%{type}"
                         hosts => "http://127.0.0.1:15100"
                }
}