centos7安装Kafka单节点环境部署三-安装Logstash

1、下载Logstash

wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.7-linux-x86_64.tar.gz

2、解压到/usr/local/

mkdir -p /usr/local/logstash7.17
tar -zxf logstash-7.17.7-linux-x86_64.tar.gz -C /usr/local/logstash7.17/ --strip-components=1 #--strip-components选项表示从目录级别上去除指定的前缀，以实现更加控制解压的效果

3、logstash数据写到kafka中

cd /usr/local/logstash7.17/config/
cp logstash-sample.conf logstash.conf
input{ 
    stdin{ }
}
output{ 
    kafka{ 
        topic_id => "logstash_k"
        bootstrap_servers => "192.168.0.191:9092"  # kafka的地址
        jaas_path => "/usr/local/logstash7.17/config/logstash.conf"
        security_protocol => "SASL_PLAINTEXT"
        sasl_mechanism => "SCRAM-SHA-256"
        codec => "json"
    }
    stdout{ 
        codec => rubydebug
    }
}

4、启动、停止

cd /usr/local/logstash7.17
mkdir -p /data/logstash7/
nohup ./bin/logstash -f config/logstash.conf &>> /data/logstash7/logstash-server-`date "+%Y%m%d"`.log & echo $! > /data/logstash7/logstash.pid #启动
cat /data/logstash7/logstash.pid | xargs -I {} kill {} #停止

logstash配置详解

Logstash 是一个强大的日志处理工具，可以用于数据收集，转换和存储。以下是 Logstash 配置文件的基本结构和一个示例配置：

# Logstash 配置文件示例
 
# 设置 Logstash 的工作区域
# path 指定了 Logstash 的工作目录
# node.name 是集群中此节点的名称
# cluster.name 是集群的名称
# pipeline.workers 是执行输入和过滤器的工作线程数
# pipeline.batch.size 是每个批次的事件数
# pipeline.batch.delay 是等待批次填满的时间
 
pipeline.workers: 2
pipeline.batch.size: 125
pipeline.batch.delay: 50
 
# 输入配置
# 使用 file 插件监控日志文件的变化
input {
  file {
    path => "/var/log/system.log"
    start_position => "beginning"
  }
}
 
# 过滤器配置
# 使用 grok 插件解析日志格式
filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message}" }
    add_field => [ "received_at", "%{@timestamp}" ]
    add_field => [ "received_from", "%{host}" ]
  }
}
 
# 输出配置
# 将处理后的事件发送到 Elasticsearch
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
  }
}

在这个配置中，我们定义了 Logstash 的工作区域和线程设置，然后配置了一个文件输入来监控系统日志文件的变化。接着，我们使用了 grok 插件来解析日志消息，并添加了两个额外字段。最后，我们配置了 Elasticsearch 作为输出，将数据发送到 Elasticsearch 索引中。