部署logstash

下载地址：https://www.elastic.co/cn/downloads/past-releases/ （注意这里是跟elasticsearch版本一致）

#创建目录
mkdir -p /home/work
#解压到指定目录
tar -xvf logstash-7.17.3-linux-x86_64.tar.gz -C /home/work

在本机上创建一个配置文件

# Kafka 输入配置，从 Kafka 主题 "sunlixin-event-log" 中获取数据
input {
  kafka {
    bootstrap_servers => "10.175.234.41:9092,10.176.19.12:9092,10.176.23.40:9092" # Kafka 服务器地址
    topics => ["sunlixin-event-log"] # 要订阅的 Kafka 主题列表
    group_id => "logstash-group-sunlixin" # Kafka 消费者组 ID
    codec => "json" # 从 Kafka 接收的数据编码格式
    decorate_events => true # 为事件添加元数据信息
    consumer_threads => 1 # 消费者线程数
  }
}

# 过滤器配置，根据 Kafka 主题名称添加索引名字段
filter {
  if [@metadata][kafka][topic] == "sunlixin-event-log" {
    mutate {
      add_field => { "index_name" => "sunlixin-event-%{+YYYY.MM.dd}" } # 添加索引名字段，格式为 "sunlixin-event-年.月.日"
    }
  }
}

# 输出配置，将处理后的数据发送至 Elasticsearch 集群
output {
  elasticsearch {
    hosts => ["10.189.113.143:9200", "10.189.116.25:9200", "10.189.116.40:9200"] # Elasticsearch 主机地址列表
    index => "%{index_name}" # 使用动态索引名，根据 filter 阶段添加的字段确定索引名
    user => "elastic" # Elasticsearch 用户名
    password => "123456" # Elasticsearch 密码
  }
}

启动

#后台启动
nohup ./logstash -f /home/work/logstash/logstash-kafka/logstash-kafka.conf &

查看进程和日志来判断是否启动成功，最好通过查看elasticsearch 使用有对应的index和数据