大数据-188 Elasticsearch - ELK 家族 Logstash Output 插件

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（已更完）
• Flink（已更完）
• ClickHouse（已更完）
• Kudu（已更完）
• Druid（已更完）
• Kylin（已更完）
• Elasticsearch（正在更新…）

章节内容

上节我们完成了如下的内容：

• Logstash Filter 插件 使用详解

Output插件

Logstash Output 插件是 Logstash 管道中的最后一个阶段，负责将处理后的数据输出到各种目标系统或存储设备。每一个输出插件都能将 Logstash 接收到并解析的数据发送到不同的外部服务，如数据库、消息队列、搜索引擎、文件系统等。

工作原理

Logstash 的输入插件（Input Plugins）从源获取数据，过滤插件（Filter Plugins）用于处理、转换数据，输出插件（Output Plugins）则是数据处理流程的最后一步。输出插件负责将 Logstash 中的事件推送到外部系统，并支持不同的输出目标，确保数据按用户需求存储或传输。

主要功能

• 多种输出目标：支持广泛的输出目标，如 Elasticsearch、Kafka、RabbitMQ、Amazon S3、文件、HTTP、数据库等。
• 批处理：在某些情况下，输出插件可以批量发送数据，而非逐条处理，从而提高性能。例如，Elasticsearch 插件可以配置批量请求，以优化写入性能。
• 数据格式化：插件能够根据需要以特定格式（如 JSON、CSV、Plaintext）输出数据。
• 重试机制：某些输出插件具有自动重试功能，确保在网络故障或系统不稳定时数据不会丢失。
• 动态路由：可以根据事件的内容动态决定数据输出到哪个目标系统。通过条件语句，可以对不同类型的数据使用不同的输出路径。
• 插件配置灵活性：每个插件都有丰富的配置选项，可以定制行为，比如指定目标主机、端口、身份验证机制、缓冲设置、批处理参数等。

高级功能

• 输出负载均衡：可以配置多个输出目标，Logstash 会自动进行负载均衡，将数据发送到不同的目标，提高数据的可用性和传输效率。
• 事件条件判断：可以使用 Logstash 的条件语句，根据事件的内容决定是否输出或选择输出的目标。例如，可以对不同的日志级别使用不同的输出目标。
• 插件组合：可以在同一个 Logstash 管道中使用多个输出插件，将数据同时发送到不同的目标，比如同时存储到 Elasticsearch 和发送到 Kafka。

常见问题

• 网络问题：由于输出插件大多与外部系统交互，网络问题可能导致输出失败。许多插件提供了重试机制和缓冲设置，以应对短暂的网络波动。
• 性能瓶颈：某些输出插件（例如 Elasticsearch、Kafka）在高并发场景下需要调整批处理和缓冲参数，以避免性能瓶颈。 数据格式问题：输出的数据格式需要与目标系统兼容，尤其是输出到数据库或 API 时，需要仔细定义数据结构和字段映射。

标准输出到控制台

将收集到的数据直接打印到控制台：

cd /opt/servers/logstash-7.3.0
bin/logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'

运行之后，对应的结果如下所示：

采集的数据保存到文件中

需求描述

Logstash也可以将收集到的数据写入到文件当中去保存，接下来我们看看Logstash如何配置以实现将数据写入到文件当中。

编写配置

cd /opt/servers/logstash-7.3.0/config
vim output_file.conf

写入的内容如下：

input {stdin{}}
output {file{
    path =>"/opt/servers/es/%{+YYYY-MM-dd}-%{host}.txt"
    codec => line {format=>"%{message}"}
    flush_interval =>0}}

写入的内容截图如下所示：

检查配置

cd /opt/servers/logstash-7.3.0
bin/logstash -f /opt/servers/logstash-7.3.0/config/output_file.conf -t 

检查结果如下图所示：

启动配置

cd /opt/servers/logstash-7.3.0
bin/logstash -f /opt/servers/logstash-7.3.0/config/output_file.conf

启动的结果如下图所示：

测试数据

我们在控制台中输入一些测试的内容，然后查看输出的文件的内容：

hello wzk icu!

查看的结果如下图所示：

查看内容：

cat /opt/servers/es/2024-08-19-h121.wzk.icu.txt

结果如下图所示：

采集数据保存到Elasticsearch

需求描述

不过多描述，主要就是把收集到的数据写入到Elasticsearch中。

编写配置

cd /opt/servers/logstash-7.3.0/config
vim output_es.conf

写入的内容如下：

input {stdin{}}
output {
  elasticsearch {
    hosts =>["h121.wzk.icu:9200"]
    index =>"logstash-%{+YYYY.MM.dd}"}}

对应的截图如下所示：

注意：这个index是保存到Elasticsearch的索引名称，如何命名是非常重要的，因为我们后续可能有某些需求做查询，所以最好带上时间。

检查配置

cd /opt/servers/logstash-7.3.0
bin/logstash -f /opt/servers/logstash-7.3.0/config/output_es.conf -t

运行的结果如下图所示：

启动配置

cd /opt/servers/logstash-7.3.0
bin/logstash -f /opt/servers/logstash-7.3.0/config/output_es.conf

启动结果如下图所示：

测试数据

向控制台写入 hello wzk icu !
查看ES集群中的数据如下图所示：