一、elastic

1.1 logstash-output-elastic

1.1.1 简单测试输入到es

vim /usr/local/logstash/1-cmdinputToEs.conf

input{
    stdin{}
}
output{
    elasticsearch{
        hosts=>[ "elk-node1:9200" ]
        index => "cmdinput-%{+YYYY.MM.dd}"
    }
   stdout{
        codec=>rubydebug
    }
}

测试

./bin/logstash -f 1-cmdinputToEs.conf

1.1.2 CODEC合并多行为一个事件

vim 2-codec-elklog.conf

input{
    file{
        path => "/var/log/elasticsearch/elk-cluster.log"
        type=> "elk"
        start_position => "beginning"
        codec => multiline {
            pattern => "^\["
            negate => "true"
            what => "previous"
        }
    }
}
output{
    if [type] == "elk" {
        elasticsearch {
            hosts => [ "elk-node1:9200" ]
            index => "elk-%{+YYYY.MM.dd}"
        }
    }
}

测试

./bin/logstash -f 2-codec-elklog.conf

二、redis

2.1 redis-input-logstash

vim redis-input-logstash.conf

input{
    redis {
        host =>"192.168.200.21"
        port =>" 6379"
        db =>"6"
        data_type =>"list"
        key="demo"
    }
}
output {
     elasticsearch {
            hosts => [ "192.168.200.21:9200" ]
            index => "redis-demo-%{+YYYY.MM.dd}"
     }
}

2.2 logstash-output-redis

vim logstash-output-redis.conf

input {
    stdin {}
}
output {
    redis {
        host =>"192.168.200.21"
        port =>" 6379"
        db =>"6"
        data_type =>"list"
        key="demo"
    }
}

三、kafka

3.1 kafka-input-logstash

请参考：kafka输入到logstash

3.2 logstash-output-kafka

注意：在logstash机器的hosts文件中一定要配置上kafka对应的IP和主机名称，否则可能会导致无法正常接收数据，但是却没有错误提示。【生产经验】

vim logstashToKafka.conf

input {
    stdin {
        codec => plain{ charset => "GB18030" }
    }
}
output {
  kafka {
    codec => plain{ charset => "GB18030" }
    bootstrap_servers => "kafka:9092"
    topic_id => "test"
  }
}

说明 ：如果logstash所处服务器的编码是GB18030，kafka所处服务器的编码是UTF-8，那么就需要指定charset