Logstash——输出（Output）

---

第6章：输出（Output）插件与目的地

Output插件定义了Logstash处理完数据后的去向。一个事件可以被发送到多个输出（例如，同时发送到Elasticsearch和标准输出用于调试），这通过条件语句来控制。本章将深入探讨最核心的输出目的地，特别是与Elasticsearch的集成，并分享在生产环境中如何确保输出阶段的可靠性与性能。

6.1 输出到Elasticsearch：最佳实践与性能调优

将Logstash与Elasticsearch结合是最经典、最强大的组合。Elasticsearch输出插件功能丰富，对其进行优化是生产部署的重中之重。

基础配置示例

output {elasticsearch {# 【必需】ES集群节点地址hosts => ["http://es-node01:9200", "http://es-node02:9200"]# 【必需】指定索引名称。支持动态变量命名，这是实现按日索引的关键。index => "app-logs-%{+YYYY.MM.dd}"# 【强烈推荐】设置一个对应用程序有意义的文档类型（ES7+中已弱化，但通常仍会设置）document_type => "_doc"# 【重要】设置数据源的唯一标识符。通常使用Filter阶段生成的唯一ID（如`fingerprint` filter）# 如果未提供，ES将自动生成一个。提供`document_id`可以实现幂等写入。# document_id => "%{fingerprint}"# 【认证】如果ES集群启用了安全特性user => "logstash_writer"password => "your_password"# 或者使用API密钥# api_key => "base64encoded_api_key"# 【SSL/TLS】启用SSL并配置CA证书ssl => truecacert => "/path/to/your/cacert.pem"}
}

性能调优与最佳实践

1. 批量写入（Bulk API）：
   Elasticsearch输出插件自动使用ES的Bulk API进行批量写入，这是性能的核心。

   elasticsearch {hosts => ["http://es01:9200"]index => "my-logs-%{+YYYY.MM.dd}"# 控制批量操作的行动（index, create, update, delete）。默认为`index`。action => "index"# 单个Bulk请求中包含的事件数。增加此值可提升吞吐，但会增加内存开销和延迟。flush_size => 10000# 发送Bulk请求的最大间隔秒数，即使未达到`flush_size`。idle_flush_time => 5
   }
   

   调优建议：flush_size通常在5000到20000之间寻找最优值。监控ES节点的负载，如果CPU和IO未打满，可以适当增加该值。idle_flush_time确保低流量数据也能被及时发送。

2. 重试与容错机制：
   Logstash内置了出色的重试机制来处理ES节点的临时故障。

   elasticsearch {hosts => ["http://es01:9200"]index => "my-logs-%{+YYYY.MM.dd}"# 发生可重试错误时（如网络问题、429状态码），重试次数。retry_initial_interval => 1retry_max_interval => 64max_retries => 10# 如果发生不可重试错误（如400 Bad Request），是否允许重试。应设为false。retry_on_conflict => false# 发生无法重试的错误后，将失败的事件写入到死信队列（DLQ），而不是丢弃。dead_letter_queue_enable => truedead_letter_queue_index => "logstash-dlq-%{+YYYY.MM.dd}"
   }
   

   建议：务必启用死信队列（DLQ）。这可以捕获因数据格式错误、映射冲突等导致无法写入ES的事件，为数据修复和重放提供可能。

3. 索引模板管理：
   为了让ES为Logstash创建的数据生成最优的映射（Mapping），应该在Logstash或ES中提前配置索引模板。

   • 方式一：让Logstash管理模板（简单）：

     elasticsearch {hosts => ["http://es01:9200"]index => "my-logs-%{+YYYY.MM.dd}"# 指定模板JSON文件路径template => "/path/to/your/logstash-template.json"template_name => "logstash-custom"template_overwrite => true
     }
     

   • 方式二：在Elasticsearch中直接管理模板（推荐用于复杂集群）：
     使用Kibana的Stack Management或ES的API直接上传和管理索引模板。这样可以实现更集中的配置管理。

4. 分布式架构：
   对于大规模部署，不要让所有Logstash节点直接写入ES主集群。

   • 推荐架构：Logstash -> Kafka -> Logstash (专用Ingest节点) -> Elasticsearch
     专门的Ingest节点只负责数据接收和写入，配置更简单，且与负责数据处理的Logstash节点解耦。

6.2 输出到标准输出stdout（用于调试）

这是一个不可或缺的调试工具，但在生产环境中应极其谨慎地使用。

output {# 在开发测试阶段，同时输出到stdout和ES，方便对比查看if "debug" in [tags] { # 使用条件语句控制，避免生产环境输出过多内容stdout {codec => rubydebug # 使用美观格式化的输出，人类可读}}# 生产环境的主要输出elasticsearch { ... }
}

警告：stdout输出会严重拖慢整个Logstash管道的性能，并产生大量不必要的磁盘I/O（如果输出到日志文件）。绝不在生产环境中默认开启，仅用于临时调试。

6.3 输出到消息队列：kafka, redis（作为缓冲或分发器）

将Logstash作为生产者，将数据写入消息队列，是实现数据缓冲、流量削峰和数据分发的核心架构模式。

输出到Kafka

output {kafka {codec => json # 通常输出JSON格式topic_id => "processed-logs-topic" # 目标主题bootstrap_servers => "kafka-broker1:9092,kafka-broker2:9092"# 可靠性配置acks => "all" # 确保消息被ISR中的所有副本确认，可靠性最高compression_type => "snappy" # 启用压缩，减少网络带宽占用batch_size => 16384 # Kafka producer的批量大小linger_ms => 5      #  producer等待批量就绪的时间}
}

应用场景：

1. 解耦：数据处理Logstash集群与数据存储/消费系统解耦。
2. 缓冲：应对下游ES集群的维护或突发流量。
3. 多路分发：一份数据写入Kafka，可供多个消费者（如ES集群、Spark流处理、审计系统）同时消费。

输出到Redis

output {redis {data_type => "list"    # 或 "channel"key => "logstash:output" # list或channel的keyhost => "redis-host"port => 6379# 如果Redis需要密码password => "your_redis_password"db => 0batch => true # 是否使用RPUSH/LPUSH批量操作batch_events => 50 # 批量大小}
}

应用场景：通常用作一个简单的、高性能的临时队列或发布/订阅通道。由于其持久化和可靠性不如Kafka，在大型关键系统中作为主要队列的场景已逐渐减少。

6.4 输出到存储系统：s3, file

输出到Amazon S3

用于数据归档和长期存储，满足合规性要求。

output {s3 {access_key_id => "your_aws_access_key"secret_access_key => "your_aws_secret_key"region => "us-east-1"bucket => "my-log-archive"# 定义S3中的对象路径和名称prefix => "logs/%{+YYYY}/%{+MM}/%{+dd}/"size_file => 10485760 # 每个文件达到10MB后滚动生成新文件time_file => 60       # 每60分钟滚动一次文件，即使未达到size_filecodec => "json_lines" # 通常以JL格式存储，便于后续分析# 还可以启用服务器端加密等# server_side_encryption => true# encoding => "gzip"    # 在写入S3前进行gzip压缩}
}

输出到本地文件

通常用于调试或特殊的数据导出需求。

output {file {path => "/path/to/output/file_%{+YYYY-MM-dd}.log"codec => json_linesgzip => true # 启用gzip压缩# 确保目录存在且Logstash进程有写权限}
}

6.5 输出到监控系统：nagios, zabbix, prometheus

Logstash可以解析日志并生成告警事件，然后直接输出到监控系统。

输出到Prometheus（通过HTTP Exposition API）

output {# 假设在filter中已计算了错误率并存入`error_rate`字段http {url => "http://prometheus-pushgateway:9091/metrics/job/logstash/instance/%{host}"http_method => "put"format => "message"message => '# TYPE app_error_rate gauge\napp_error_rate{host="%{host}"} %{error_rate}'content_type => "text/plain"}
}

注意：更常见的做法是使用Prometheus的Node Exporter或Exporters来抓取指标，或者使用metricbeat来收集Logstash自身的指标。HTTP输出通常用于推送到Pushgateway的特殊场景。

6.6 输出到其他网络服务：tcp, http

作为通用的数据发送器，将数据推送到任何开放的TCP Socket或HTTP端点。

tcp 输出

output {tcp {host => "destination-host"port => 9000codec => json_lines # 将事件编码为JSON Lines格式通过网络发送}
}

http 输出

output {http {url => "https://my-external-api.com/ingest"http_method => "post"format => "json" # 将整个事件作为JSON body发送content_type => "application/json"# 可以添加认证头headers => ["Authorization", "Bearer your_api_token"]# 重试配置retry_failed => trueretry_non_idempotent => trueretryable_codes => [408, 429, 500, 502, 503, 504]}
}

应用场景：将处理后的数据发送到自定义的API、第三方SaaS服务（如Slack、PagerDuty用于告警）、或另一个数据处理系统。

总结

Output插件定义了数据的最终价值实现。：

1. 目的地选择：是直接存储（ES/S3），还是缓冲分发（Kafka），或是触发动作（HTTP告警）？
2. 可靠性设计：如何配置重试、死信队列和批量参数来保证数据不丢失？
3. 性能优化：针对不同的输出，如何调整批量大小、并发度和压缩来最大化吞吐量？
4. 条件输出：熟练运用if [field]条件语句，将不同的事件路由到不同的输出，实现复杂的数据流编排。