tfl-cycle-journey-pipeline.conf

input {
    file {
        mode => "read"
        path => "${CYCLE_JOURNEY_CSV_PATH}"
    }
}

filter {
    csv {
        remove_field => ["message"]
    }

    if [column1] == "Rental Id" {
        drop {}
    }

    date {
        match => ["column4", "dd/mm/yyyy HH:mm"]
        target => "EndDate"
        remove_field => ["column4"]
    }

    date {
        match => ["column7", "dd/mm/yyyy HH:mm"]
        target => "StartDate"
        remove_field => ["column7"]
    }

    mutate {
        rename => {
            "column1" => "RentalId"
            "column2" => "Duration"
            "column3" => "BikeId"
            "column5" => "EndStationId"
            "column6" => "EndStationName"
            "column8" => "StartStationId"
            "column9" => "StartStationName"
        }
        copy => {
            "StartDate" => "@timestamp"
        }
    }
}

output {
    elasticsearch {
        hosts => ["http://10.0.200.101:9200"]
        index => "cycle-journeys"
        document_id => "%{RentalId}"
    }
}