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/ha1?zeroDateTimeBehavior=convertToNull&mha1?zeroDateTimeBehavior=convertToNull"
    jdbc_user => "root"
    jdbc_password => "root"
    schedule => "* * * * * "
    statement => "select * from `su_ha1`.`cts_event_control` AS e
JOIN
`su_ha1`.`cts_event_control_cstm` AS c ON c.id_c = e.id
LEFT JOIN
`ha1`.`events` ctsE ON c.event_id_c = ctsE.event_id
LEFT JOIN
`ha1`.`series` ctsS ON c.series_id_c = ctsS.series_id"
    type => "series_events"
  }
  # add more jdbc inputs to suit your needs
}
filter {

date {
    match => [ "[evtlist_date_c]", "yyyy-MM-dd" ]
    target => "evtlist_date_c"
  }

  if "_dateparsefailure" in [tags] {
    mutate {
      replace => { "evtlist_date_c" => "1970-01-01" }
    }
	date{
      match => [ "[evtlist_date_c]", "yyyy-MM-dd" ]
      target => "evtlist_date_c"
    }
  }



}
output {
               stdout { codec => rubydebug {} }
}