2+1

pipeline有两个必选元素：input and output，一个可选：Filter：

conf

input {
    kafka {
        bootstrap_servers => "192.168.40.148:9092"
        group_id => "logstash-es"
        topics => ["es"]
        decorate_events => false
        codec => json
        client_id => "node-es"
    }

}

filter { 
    mutate{
        id => "es_add_pipeline"
        add_field => {
            "pipeline" => "es"
        }
    }
}

output { 
        elasticsearch {
            action => "index"
            hosts => ["192.168.40.148:9200"]
            index  => "topic-%{pipeline}-%{+YYYY.MM.dd}"
        }
}