output

yt-summer

output区块为事件输出目的地，常用插件有stdout，elasticsearch，kafka等
更多output插件知识查看官网文档：https://www.elastic.co/guide/en/logstash/current/output-plugins.html

stdout插件：
一个简单的打印输出，多用来进行debug
示例：
output {
stdout {
codec => rubydebug
workers => 2
}
}

workers:多线程输出
rubydebug使用ruby的Awesome打印库打印事件输出


elasticsearch插件：
使用http协议将事件写入elasticsearch
示例：

output {

elasticsearch {

hosts => ["172.16.133.129:9200", "172.16.133.133:9200"]

index => "logstash-%{type}-%{+YYYY.MM.dd}"

workers => 2

flush_size => 20000

idle_flush_time => 10

}

}


hosts:连接elasticsearch服务器
index:写入elasticsearch的索引名。这里可以使用变量。为了更贴合日志场景，Logstash 提供了 %{+YYYY.MM.dd} 这种写法。在语法解析的时候，看到以 + 号开头的，就会自动认为后面是时间格式，尝试用时间格式来解析后续字符串。所以，之前处理过程中不要给自定义字段取个加号开头的名字。此外，注意索引名中不能有大写字母，否则 ES 在日志中会报 InvalidIndexNameException，但是 Logstash 不会报错，这个错误比较隐晦。
flush_size 和 idle_flush_time 共同控制 Logstash 向 Elasticsearch 发送批量数据的行为。以上面示例来说：Logstash 会努力攒到 20000 条数据一次性发送出去，但是如果 10 秒钟内也没攒够 20000 条，Logstash 还是会以当前攒到的数据量发一次。默认情况下，flush_size 是 500 条，idle_flush_time 是 1 秒。这也是很多人改大了 flush_size也没能提高写入 ES 性能的原因——Logstash 还是 1 秒钟发送一次。
Logstash 在有多个 conf 文件的情况下，进入 ES 的数据会重复，几个 conf 数据就会重复几次，这是因为output 段顺序执行，没有对日志进行type判断的各插件配置都会全部执行一次。所以如果有多个conf文件，在 output 段对 type 进行判断，语法如下所示：

output {

if [type] == "test" {

elasticsearch { }

}

}

kafka插件：
将事件写入kafka的topic中
示例：

output {

if [type] == "test" {

kafka {

bootstrap_servers => "10.10.10.6:9092,10.10.10.8:9092,10.10.10.9:9092"

topic_id => "command_audit_log"

compression_type => "snappy"

}

}

}


bootstrap_servers:kafka server的地址
topic_id:生成消息的话题(必填项)
compression_type:压缩类型(snappy,gzip,none)
该插件默认的codec为json，所以事件将以json的格式导入，如果配置codec为plain，logstash除了编码messages外，还会添加timestamp和hostname的字段，如果只想输出message信息，配置如下output：

output {

kafka {

        codec => plain {

format => "%{message}"

}

}

}