目录一.ES数据备份与恢复1.1简单介绍1.2实现数据备份1.2.1配置快照仓库——nfs1.2.2配置ES集群节点1.2.3创建API仓库1.2.4创建快照1.2.5恢复快照1.2.6删除快照1.2.7ES集群管理界面管理快照二.ES集群监控2.1简单介绍2.2启用xpack认证2.2.2修改elasticsearch.yml文件并重启服务2.2.3将生成的证书远程传输给另外两台主机2.2.4完成server2和server3上的配置2.2.5为EFLK中的各个组件设置密码2.2.6访问Cerebro2.2.7访问Logstash2.2.8访问Kibana2.3实现metricbeat监控2.3.1简单介绍2.3.2进入KIbana的内部监测开启Metricbeat2.3.3三台主机安装metricbeat软件包并启动模块2.3.4三台主机编辑elasticsearch-xpack.yml文件2.3.5三台主机修改metricbeat.yml文件2.3.6三台主机重新进入Kibana查看是否配置成功2.4kibana 监控2.4.1三台主机安装filebeat软件包并启动elasticsearch模块2.4.2三台主机编辑elasticsearch.yml文件2.4.3三台主机编辑filebeat.yml文件2.4.4查看日志能否监测到三.ELFKkafka架构3.1简单介绍3.2部署ELFKkafka架构3.2.1准备工作3.2.2三台主机安装 JDK 环境3.3.3三台主机安装 Kafka3.3.4三台主机 Kafka 配置文件修改3.3.5Kafka 集群初始化3.3.6Kafka 集群可用性验证3.3修改FIlebeat 配置文件3.3.1修改filebeat.yml文件3.3.2Kafka 集群任意节点消费者测试3.4Logstash 消费 Topic 消息一.ES数据备份与恢复1.1简单介绍1.说明ES 数据备份与恢复基于“快照 仓库”机制通过将索引数据以增量方式备份到本地文件系统或云存储中实现灾难恢复、数据迁移和历史归档能力。通过 快照Snapshot 和 恢复Restore 机制实现数据备份依赖于共享存储仓库如本地文件系统、S3、HDFS 等1.2实现数据备份说明在server4上配置快照仓库接着在ES集群中每个主机上安装配置nfs指明快照的存储路径最后在ES集群管理界面上交互生成快照和恢复快照或者进入kibana生成快照和恢复快照1.2.1配置快照仓库——nfs说明NFS网络文件系统是一种分布式文件系统协议它允许你在网络上像访问本地文件一样访问远程服务器上的文件和目录。1.安装nfs软件yum install -y nfs-utils2.创建一个目录用于存储拍摄下的ES集群快照mkdir -p /data/es-backup chmod 777 /data/es-backup3.编辑/etc/exports文件并启动nfs服务vim /etc/exports systemctl enable --now nfs1.2.2配置ES集群节点1.安装nfs软件包yum install -y nfs-utils2.创建一个与server4路径相同的路径用于保存同时将Es集群的主机远程挂载到nfs的server端mkdir -p /data/es-backup mount 192.168.7.164:/data/es-backup #远程挂载3.修改estaicsearch.yml文件将快照保存到远程server4的目录下vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch1.2.3创建API仓库1.进入Kibana控制台输入以下指令创建仓库PUT /_snapshot/my_backup # 仓库名称自定义 { type: fs, # 本地文件系统类型 settings: { location: /data/es-backup, # 与elasticsearch.yml中配置一致 compress: true # 启用压缩可选 } }2.验证仓库是否创建成功GET /_snapshot/my_backup1.2.4创建快照1.定义快照名称PUT /_snapshot/my_backup/snapshot_all_2026.08.11 # 快照名称自定义建议含时间2.备份指定索引PUT /_snapshot/my_backup/snapshot_syslog_2026.08.11 { indices: syslog-*, # 逗号分隔的索引名或通配符 ignore_unavailable: true, # 忽略不存在的索引 include_global_state: false # 是否备份集群全局状态如模板、别名等 }3.列出仓库中所有快照GET /_snapshot/my_backup/_all4.查看单个快照的详细情况GET /_snapshot/my_backup/snapshot_syslog_2026.08.111.2.5恢复快照1.将快照中的所有索引恢复到集群若索引已存在默认不覆盖需指定rename_pattern重命名POST /_snapshot/my_backup/snapshot_syslog_2026.08.11/_restore2.仅恢复快照中的部分索引并可重命名避免冲突POST /_snapshot/my_backup/snapshot_all_20251026/_restore { indices: syslog-2026.08.11, # 仅恢复syslog-2025.10.26索引 rename_pattern: syslog-(\\d), # 匹配原索引名的正则 rename_replacement: restored_syslog-$1, # 重命名 include_global_state: false # 不恢复集群全局状态 }1.2.6删除快照DELETE /_snapshot/my_backup/snapshot_all_2026.08.111.2.7ES集群管理界面管理快照二.ES集群监控2.1简单介绍1.简单说明ES集群监控是通过实时采集集群健康状态、性能指标如查询延迟、索引速率和资源水位CPU、内存、磁盘确保集群稳定运行并提前发现故障隐患的一套可观测性体系。2.2启用xpack认证2.2.1在server1上生成证书cd /usr/share/elasticsearch/ bin/elasticsearch-certutil ca bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12 cp elastic-certificates.p12 /etc/elasticsearch cd /etc/elasticsearch/ chown elasticsearch elastic-certificates.p122.2.2修改elasticsearch.yml文件并重启服务vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch2.2.3将生成的证书远程传输给另外两台主机scp /etc/elassticsearch/elastic-certificates.p12 server2:/etc/elasticsearch/ scp /etc/elassticsearch/elastic-certificates.p12 server3:/etc/elasticsearch/2.2.4完成server2和server3上的配置chown elasticsearch /etc/elasticsearch/elastic-certificates.p12 vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch.service2.2.5为EFLK中的各个组件设置密码说明这里只需要ES集群中的一台主机设置即可他会自动同步到其他节点上cd /usr/share/elasticsearch/ bin/elasticsearch-setup-passwords interactive2.2.6访问Cerebro说明设置了E在访问EFLK中的任意一个组件时需要在url中输入用户和密码cd elasticsearch-head-master/ npm run start docker start cerebro #启动cerebro2.2.7访问Logstashvim /etc/logstash/conf.d/grok.conf /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf2.2.8访问Kibanavim /etc/kibana/kibana.yml systemctl restart kibana.service netstat -antlp|grep :56012.3实现metricbeat监控2.3.1简单介绍1.说明Metricbeat是一个轻量型的指标采集器通过启用其Elasticsearch模块可以主动从ES集群拉取节点、分片、索引等各项指标数据并将其输送至Elasticsearch或Kibana中进行存储和可视化展示2.3.2进入KIbana的内部监测开启Metricbeat注意ES集群中的所有主机都要完成下列操作2.3.3三台主机安装metricbeat软件包并启动模块说明elasticsearch-xpack模块用于开启Metricbeat对Elasticsearch集群的性能指标采集并支持基于X-Pack的安全认证与加密连接。rpm -ivh metricbeat-7.6.1-x86_64.rpm cd /etc/metricbeat/modules.d metricbeat modules enable elasticsearch-xpack2.3.4三台主机编辑elasticsearch-xpack.yml文件vim elasticsearch-xpack.yml2.3.5三台主机修改metricbeat.yml文件cd /etc/metricbeat vim metricbeat.yml systemctl enable --now metricbeat.service2.3.6三台主机重新进入Kibana查看是否配置成功注意ES集群中的所有机子都完成操作后查看2.4kibana 监控1.说明ES集群中的所有机子都要实现下列步骤2.4.1三台主机安装filebeat软件包并启动elasticsearch模块rpm -ivh filebeat-7.6.1-x86_64.rpm cd /etc/filebeat/modules.d filebeat modules enable elasticsearch2.4.2三台主机编辑elasticsearch.yml文件vim /etc/filebeat/modules.d/elasticsearch.yml2.4.3三台主机编辑filebeat.yml文件vim /etc/filebeat/modules.d/filebeat.yml systemctl enable --now filebeat2.4.4查看日志能否监测到三.ELFKkafka架构3.1简单介绍1.说明ELFKKafka架构是通过引入Kafka消息队列作为缓冲层将日志采集Filebeat与处理Logstash解耦实现高并发场景下的流量削峰填谷保障日志管道不丢失、不阻塞。2.KafkaKafka是一个分布式、高吞吐量的消息队列系统用于在系统之间可靠地传递数据流起到缓冲和解耦的作用。3.2部署ELFKkafka架构3.2.1准备工作1.准备8台CentOS 7虚拟机主机名IP地址角色server1192.168.7.171elasticsearch,filebeatserver2192.168.7.172elasticsearch,filebeatserver3192.168.7.173elasticsearch,filebeatserver4192.168.7.174cerebroserver5192.168.7.175Logstash,Kabanaserver6192.168.7.176Kafkaserver7192.168.7.177Kafkaserver8192.168.7.178Kafka2.完成各个主机上的域名解析3.2.2三台主机安装 JDK 环境rpm -ivh jdk-17.0.17_linux-x64_bin.rpm java -version3.3.3三台主机安装 Kafka官方安装包链接https://kafka.apache.org/community/downloads/https://kafka.apache.org/downloadstar zxf kafka_2.12-3.9.1.tgz -C /opt mv /opt/kafka_2.12-3.9.1 /opt/kafka #重命名 cd /opt/kafka/config/kraft cp server.properties{,.bak} # 备份配置文件 mkdir /opt/kafka/logs # 创建日志目录3.3.4三台主机 Kafka 配置文件修改vim /opt/kraft/server.properties scp /opt/kraft/server.properties server7:/opt/kraft/server.properties scp /opt/kraft/server.properties server8:/opt/kraft/server.properties3.3.5Kafka 集群初始化1.在server6 节点生成储目录唯一的 UUIDKAFKA_CLUSTER_ID$(/opt/kafka/bin/kafka-storage.sh random-uuid) echo ${KAFKA_CLUSTER_ID}2.三台主机用该 uuid 格式化 kafka 存储目录/opt/kafka/bin/kafka-storage.sh format -t 你的UUID -c /opt/kafka/config/kraft/server.properties3.三台节点启动 Kafka/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties jps3.3.6Kafka 集群可用性验证1.Kafka 集群任意节点创建 Topic/opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3--replication-factor 3设置主题的副本因子为 3即每个分区的副本数量为 3。副本用于数据冗余和高可用性副本数需不能超过集群中 Broker 的总数。--partitions 3设置主题的分区数为 3。分区是 Kafka 并行处理消息的基本单位影响吞吐量和扩展性。分区数一旦创建不可减少。2.选择Kafka 集群的其他节点查看Topic是否创建/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:90923.Kafka 集群任意节点生产者测试/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopics4.Kafka 集群任意节点查看主题/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginning5.Kafka 集群任意节点查看 topic 详细信息/opt/kafka/bin/kafka-topics.sh --bootstrap-server server6:9092,server7:9092,server8:9092 --topic testtopics --describe3.3修改FIlebeat 配置文件3.3.1修改filebeat.yml文件vim /etc/filebeat/filebeat.yml filebeat test config -c /etc/filebeat/filebeat.yml #检验文件语法 filebeat test output -c /etc/filebeat/filebeat.yml #测试文件 systemctl restart filebeat3.3.2Kafka 集群任意节点消费者测试/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginning3.4Logstash 消费 Topic 消息说明进入server5到/etc/logstash/conf.d目录下创建一个kafka.conf配置文件cd /etc/logstash/conf.d vim kafka.conf /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf
目录一.ES数据备份与恢复1.1简单介绍1.2实现数据备份1.2.1配置快照仓库——nfs1.2.2配置ES集群节点1.2.3创建API仓库1.2.4创建快照1.2.5恢复快照1.2.6删除快照1.2.7ES集群管理界面管理快照二.ES集群监控2.1简单介绍2.2启用xpack认证2.2.2修改elasticsearch.yml文件并重启服务2.2.3将生成的证书远程传输给另外两台主机2.2.4完成server2和server3上的配置2.2.5为EFLK中的各个组件设置密码2.2.6访问Cerebro2.2.7访问Logstash2.2.8访问Kibana2.3实现metricbeat监控2.3.1简单介绍2.3.2进入KIbana的内部监测开启Metricbeat2.3.3三台主机安装metricbeat软件包并启动模块2.3.4三台主机编辑elasticsearch-xpack.yml文件2.3.5三台主机修改metricbeat.yml文件2.3.6三台主机重新进入Kibana查看是否配置成功2.4kibana 监控2.4.1三台主机安装filebeat软件包并启动elasticsearch模块2.4.2三台主机编辑elasticsearch.yml文件2.4.3三台主机编辑filebeat.yml文件2.4.4查看日志能否监测到三.ELFKkafka架构3.1简单介绍3.2部署ELFKkafka架构3.2.1准备工作3.2.2三台主机安装 JDK 环境3.3.3三台主机安装 Kafka3.3.4三台主机 Kafka 配置文件修改3.3.5Kafka 集群初始化3.3.6Kafka 集群可用性验证3.3修改FIlebeat 配置文件3.3.1修改filebeat.yml文件3.3.2Kafka 集群任意节点消费者测试3.4Logstash 消费 Topic 消息一.ES数据备份与恢复1.1简单介绍1.说明ES 数据备份与恢复基于“快照 仓库”机制通过将索引数据以增量方式备份到本地文件系统或云存储中实现灾难恢复、数据迁移和历史归档能力。通过 快照Snapshot 和 恢复Restore 机制实现数据备份依赖于共享存储仓库如本地文件系统、S3、HDFS 等1.2实现数据备份说明在server4上配置快照仓库接着在ES集群中每个主机上安装配置nfs指明快照的存储路径最后在ES集群管理界面上交互生成快照和恢复快照或者进入kibana生成快照和恢复快照1.2.1配置快照仓库——nfs说明NFS网络文件系统是一种分布式文件系统协议它允许你在网络上像访问本地文件一样访问远程服务器上的文件和目录。1.安装nfs软件yum install -y nfs-utils2.创建一个目录用于存储拍摄下的ES集群快照mkdir -p /data/es-backup chmod 777 /data/es-backup3.编辑/etc/exports文件并启动nfs服务vim /etc/exports systemctl enable --now nfs1.2.2配置ES集群节点1.安装nfs软件包yum install -y nfs-utils2.创建一个与server4路径相同的路径用于保存同时将Es集群的主机远程挂载到nfs的server端mkdir -p /data/es-backup mount 192.168.7.164:/data/es-backup #远程挂载3.修改estaicsearch.yml文件将快照保存到远程server4的目录下vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch1.2.3创建API仓库1.进入Kibana控制台输入以下指令创建仓库PUT /_snapshot/my_backup # 仓库名称自定义 { type: fs, # 本地文件系统类型 settings: { location: /data/es-backup, # 与elasticsearch.yml中配置一致 compress: true # 启用压缩可选 } }2.验证仓库是否创建成功GET /_snapshot/my_backup1.2.4创建快照1.定义快照名称PUT /_snapshot/my_backup/snapshot_all_2026.08.11 # 快照名称自定义建议含时间2.备份指定索引PUT /_snapshot/my_backup/snapshot_syslog_2026.08.11 { indices: syslog-*, # 逗号分隔的索引名或通配符 ignore_unavailable: true, # 忽略不存在的索引 include_global_state: false # 是否备份集群全局状态如模板、别名等 }3.列出仓库中所有快照GET /_snapshot/my_backup/_all4.查看单个快照的详细情况GET /_snapshot/my_backup/snapshot_syslog_2026.08.111.2.5恢复快照1.将快照中的所有索引恢复到集群若索引已存在默认不覆盖需指定rename_pattern重命名POST /_snapshot/my_backup/snapshot_syslog_2026.08.11/_restore2.仅恢复快照中的部分索引并可重命名避免冲突POST /_snapshot/my_backup/snapshot_all_20251026/_restore { indices: syslog-2026.08.11, # 仅恢复syslog-2025.10.26索引 rename_pattern: syslog-(\\d), # 匹配原索引名的正则 rename_replacement: restored_syslog-$1, # 重命名 include_global_state: false # 不恢复集群全局状态 }1.2.6删除快照DELETE /_snapshot/my_backup/snapshot_all_2026.08.111.2.7ES集群管理界面管理快照二.ES集群监控2.1简单介绍1.简单说明ES集群监控是通过实时采集集群健康状态、性能指标如查询延迟、索引速率和资源水位CPU、内存、磁盘确保集群稳定运行并提前发现故障隐患的一套可观测性体系。2.2启用xpack认证2.2.1在server1上生成证书cd /usr/share/elasticsearch/ bin/elasticsearch-certutil ca bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12 cp elastic-certificates.p12 /etc/elasticsearch cd /etc/elasticsearch/ chown elasticsearch elastic-certificates.p122.2.2修改elasticsearch.yml文件并重启服务vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch2.2.3将生成的证书远程传输给另外两台主机scp /etc/elassticsearch/elastic-certificates.p12 server2:/etc/elasticsearch/ scp /etc/elassticsearch/elastic-certificates.p12 server3:/etc/elasticsearch/2.2.4完成server2和server3上的配置chown elasticsearch /etc/elasticsearch/elastic-certificates.p12 vim /etc/elasticsearch/elasticsearch.yml systemctl restart elasticsearch.service2.2.5为EFLK中的各个组件设置密码说明这里只需要ES集群中的一台主机设置即可他会自动同步到其他节点上cd /usr/share/elasticsearch/ bin/elasticsearch-setup-passwords interactive2.2.6访问Cerebro说明设置了E在访问EFLK中的任意一个组件时需要在url中输入用户和密码cd elasticsearch-head-master/ npm run start docker start cerebro #启动cerebro2.2.7访问Logstashvim /etc/logstash/conf.d/grok.conf /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf2.2.8访问Kibanavim /etc/kibana/kibana.yml systemctl restart kibana.service netstat -antlp|grep :56012.3实现metricbeat监控2.3.1简单介绍1.说明Metricbeat是一个轻量型的指标采集器通过启用其Elasticsearch模块可以主动从ES集群拉取节点、分片、索引等各项指标数据并将其输送至Elasticsearch或Kibana中进行存储和可视化展示2.3.2进入KIbana的内部监测开启Metricbeat注意ES集群中的所有主机都要完成下列操作2.3.3三台主机安装metricbeat软件包并启动模块说明elasticsearch-xpack模块用于开启Metricbeat对Elasticsearch集群的性能指标采集并支持基于X-Pack的安全认证与加密连接。rpm -ivh metricbeat-7.6.1-x86_64.rpm cd /etc/metricbeat/modules.d metricbeat modules enable elasticsearch-xpack2.3.4三台主机编辑elasticsearch-xpack.yml文件vim elasticsearch-xpack.yml2.3.5三台主机修改metricbeat.yml文件cd /etc/metricbeat vim metricbeat.yml systemctl enable --now metricbeat.service2.3.6三台主机重新进入Kibana查看是否配置成功注意ES集群中的所有机子都完成操作后查看2.4kibana 监控1.说明ES集群中的所有机子都要实现下列步骤2.4.1三台主机安装filebeat软件包并启动elasticsearch模块rpm -ivh filebeat-7.6.1-x86_64.rpm cd /etc/filebeat/modules.d filebeat modules enable elasticsearch2.4.2三台主机编辑elasticsearch.yml文件vim /etc/filebeat/modules.d/elasticsearch.yml2.4.3三台主机编辑filebeat.yml文件vim /etc/filebeat/modules.d/filebeat.yml systemctl enable --now filebeat2.4.4查看日志能否监测到三.ELFKkafka架构3.1简单介绍1.说明ELFKKafka架构是通过引入Kafka消息队列作为缓冲层将日志采集Filebeat与处理Logstash解耦实现高并发场景下的流量削峰填谷保障日志管道不丢失、不阻塞。2.KafkaKafka是一个分布式、高吞吐量的消息队列系统用于在系统之间可靠地传递数据流起到缓冲和解耦的作用。3.2部署ELFKkafka架构3.2.1准备工作1.准备8台CentOS 7虚拟机主机名IP地址角色server1192.168.7.171elasticsearch,filebeatserver2192.168.7.172elasticsearch,filebeatserver3192.168.7.173elasticsearch,filebeatserver4192.168.7.174cerebroserver5192.168.7.175Logstash,Kabanaserver6192.168.7.176Kafkaserver7192.168.7.177Kafkaserver8192.168.7.178Kafka2.完成各个主机上的域名解析3.2.2三台主机安装 JDK 环境rpm -ivh jdk-17.0.17_linux-x64_bin.rpm java -version3.3.3三台主机安装 Kafka官方安装包链接https://kafka.apache.org/community/downloads/https://kafka.apache.org/downloadstar zxf kafka_2.12-3.9.1.tgz -C /opt mv /opt/kafka_2.12-3.9.1 /opt/kafka #重命名 cd /opt/kafka/config/kraft cp server.properties{,.bak} # 备份配置文件 mkdir /opt/kafka/logs # 创建日志目录3.3.4三台主机 Kafka 配置文件修改vim /opt/kraft/server.properties scp /opt/kraft/server.properties server7:/opt/kraft/server.properties scp /opt/kraft/server.properties server8:/opt/kraft/server.properties3.3.5Kafka 集群初始化1.在server6 节点生成储目录唯一的 UUIDKAFKA_CLUSTER_ID$(/opt/kafka/bin/kafka-storage.sh random-uuid) echo ${KAFKA_CLUSTER_ID}2.三台主机用该 uuid 格式化 kafka 存储目录/opt/kafka/bin/kafka-storage.sh format -t 你的UUID -c /opt/kafka/config/kraft/server.properties3.三台节点启动 Kafka/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties jps3.3.6Kafka 集群可用性验证1.Kafka 集群任意节点创建 Topic/opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3--replication-factor 3设置主题的副本因子为 3即每个分区的副本数量为 3。副本用于数据冗余和高可用性副本数需不能超过集群中 Broker 的总数。--partitions 3设置主题的分区数为 3。分区是 Kafka 并行处理消息的基本单位影响吞吐量和扩展性。分区数一旦创建不可减少。2.选择Kafka 集群的其他节点查看Topic是否创建/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:90923.Kafka 集群任意节点生产者测试/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopics4.Kafka 集群任意节点查看主题/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginning5.Kafka 集群任意节点查看 topic 详细信息/opt/kafka/bin/kafka-topics.sh --bootstrap-server server6:9092,server7:9092,server8:9092 --topic testtopics --describe3.3修改FIlebeat 配置文件3.3.1修改filebeat.yml文件vim /etc/filebeat/filebeat.yml filebeat test config -c /etc/filebeat/filebeat.yml #检验文件语法 filebeat test output -c /etc/filebeat/filebeat.yml #测试文件 systemctl restart filebeat3.3.2Kafka 集群任意节点消费者测试/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginning3.4Logstash 消费 Topic 消息说明进入server5到/etc/logstash/conf.d目录下创建一个kafka.conf配置文件cd /etc/logstash/conf.d vim kafka.conf /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf
