基于Kafka（Kraft）+ELK的日志收集与分析平台的搭建（下）

一、ELK 三大组件介绍以及整体工作原理ELK 是一套开源日志分析技术栈由 Elasticsearch、Logstash、Kibana 三个组件组成。1.1 三大组件各自作用Elasticsearch分布式全文检索引擎。底层基于 Lucene 实现。主要负责日志 / 文档数据的存储、索引、检索、聚合统计。接收上游采集组件推送过来的数据对外提供 RESTful HTTP 接口。Logstash日志数据处理服务。主要承担数据采集、过滤、清洗、格式转换支持多种输入输出源可以对原始杂乱日志做字段提取、格式转换后输出给 Elasticsearch。Kibana可视化 Web 平台。对接 Elasticsearch提供网页界面完成索引查看、数据查询、图表可视化内置开发工具可以直接执行 ES 的 DSL 语句。1.2 ELK 标准工作流程标准完整链路业务服务器日志 → Logstash 采集清洗 → Elasticsearch 存储索引 → Kibana 做可视化查询展示1.3 Elasticsearch 底层核心工作原理1.3.1 倒排索引关系型数据库MySQL使用正排索引以文档行作为核心查询时遍历行匹配内容擅长精准查询海量文本模糊检索性能差。Elasticsearch使用倒排索引对文本内容进行分词构建「关键词 → 文档 ID」的映射关系。搜索时直接根据关键词定位对应文档实现海量数据毫秒级检索这是 ES 搜索速度快的根本原因。1.3.2 分片与副本主分片Primary Shard将索引的数据水平拆分解决单节点磁盘上限实现分布式存储主分片数量在创建索引后不可修改。副本分片Replica Shard主分片的数据完整备份。一方面实现高可用主分片故障副本可以升级为主分片另一方面分担查询压力。副本不能和对应的主分片部署在同一台机器。1.3.3 写入基础机制ES 写入并不是实时可见属于近实时 (NRT)。数据先写入内存缓冲区经过 refresh默认 1 秒生成只读 segment 段此时数据可以被检索translog 事务日志保障宕机不丢失数据后续 flush 将数据持久化磁盘后台会自动执行段合并 merge把大量小 segment 合并为大段清理标记删除的数据释放磁盘空间。1.3.4 Elasticsearch 与 MySQL 概念对比ElasticsearchMySQL 关系型数据库说明Index索引Table数据表存放同一类业务数据的集合Document文档Row数据行ES 最小数据单元以 JSON 格式存储一条记录Field字段Column列文档中的属性对应数据表的列Mapping映射Schema表结构定义字段类型、是否分词、是否建立索引等元数据_id主键Primary Key文档的唯一标识符可手动指定或由系统自动生成二、为什么选用 Elasticsearch Kibana本次不使用 Logstash2.1传统排查日志方式登录多台业务服务器使用 tail -f、grep 命令检索日志。存在痛点多机器日志分散需要登录多台服务器排查问题效率低下很难对大量日志做统计、过滤、图表分析不方便做历史日志回溯检索2.2Elasticsearch Kibana 组合带来的价值Elasticsearch 负责存储日志利用倒排索引支持海量日志快速检索、条件过滤、聚合统计Kibana 提供 Web 可视化页面不用登录服务器浏览器即可查询日志、绘制图表Logstash 功能强大但是资源消耗高本次为学习测试环境暂时舍弃 Logstash先体验 ES 存储检索和 Kibana 可视化后续生产环境再引入 Logstash 或者 Filebeat 完成日志采集清洗⚠️重要约束Elasticsearch 和 Kibana 版本必须完全一致版本不一致会出现连接报错、启动失败。本文统一使用 7.17.18 版本。三、部署环境准备3.1 基础环境操作系统CentOS10部署方式tar 二进制包部署内存建议2G 以上3.2 系统前置调优ES 运行必须配置Elasticsearch 会大量打开文件句柄使用操作系统内存映射同时出于安全策略禁止 root 用户启动需要修改系统资源限制、内核参数创建普通运行用户。3.2.1 修改文件句柄、进程数限制vim /etc/security/limits.conf追加下面内容* soft nofile 65535 * hard nofile 65535 * soft nproc 4096 * hard nproc 4096作用调高进程最大打开文件数量、最大进程数防止启动报文件描述符不足。3.2.2 修改内核虚拟内存参数vim /etc/sysctl.conf添加配置vm.max_map_count655300 fs.file-max655350配置生效sysctl -p作用vm.max_map_count 控制内存映射区域数量ES 大量使用 mmap 内存映射参数不足会直接启动失败。3.2.3 创建 es 普通用户组和用户ES 禁止 root 账号运行必须使用普通用户启动。groupadd es useradd -g es es四、服务部署Elasticsearchkibana部署在一台机器上Elasticsearch-head不是必需的4.1 Elasticsearch 7.17.18 部署4.1.1 下载和解压cd /usr/local/src wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.18-linux-x86_64.tar.gztar -zxvf elasticsearch-7.17.18-linux-x86_64.tar.gz mv elasticsearch-7.17.18 /usr/local/elasticsearch chown -R es:es /usr/local/elasticsearch说明移动到 /usr/local 统一管理程序目录修改目录归属保证 es 用户拥有读写权限。4.1.2 修改配置文件 elasticsearch.ymlvim /usr/local/elasticsearch/config/elasticsearch.ymlcluster.name: my-elasticsearch node.name: node-1 network.host: 0.0.0.0 http.port: 9200 discovery.type: single-node http.cors.enabled: true http.cors.allow-origin: *参数说明cluster.name集群名称单机自定义集群环境所有节点必须保持一致node.name节点名称集群环境每个节点名字唯一network.host: 0.0.0.0允许外部 IP 访问服务默认仅本机 127.0.0.1 访问http.port:9200ES 对外服务端口discovery.type: single-node单机模式不需要节点互相发现cors 跨域配置用于 Kibana、Head 插件访问4.1.3 启动与验证su es cd /usr/local/elasticsearch/bin前台启动适合观察日志调试./elasticsearch后台守护进程启动# ./elasticsearch -d新开终端验证服务curl http://127.0.0.1:9200返回 JSON 集群信息代表 Elasticsearch 启动成功。4.1.4 安装 IK 中文分词器ES 原生对中文分词效果很差会按单个汉字拆分IK 分词器提供专门的中文分词能力。⚠️分词器版本必须和 ES 版本完全一致。su es cd /usr/local/elasticsearch/plugins mkdir ik cd ikwget https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.17.18/elasticsearch-analysis-ik-7.17.18.zip yum install unzip -y unzip elasticsearch-analysis-ik-7.17.18.zip rm -rf elasticsearch-analysis-ik-7.17.18.zip重启 ES 插件生效ps -ef | grep elasticsearch | grep -v grep | awk {print $2} | xargs kill -9 /usr/local/elasticsearch/bin/elasticsearch -d4.1.5 部署 Elasticsearch‑Head 可视化工具作用Elasticsearch‑Head 是第三方图形化管理界面直观查看集群健康状态、索引、分片、文档数据方便调试。注意ES7.x 不再支持作为内置插件安装需要独立部署依赖 Node.js 环境前面已经部署 node‑v14.21.3。部署步骤cd /usr/local/src # 下载 head 源码 wget https://github.com/mobz/elasticsearch-head/archive/master.zip yum install unzip -y unzip master.zip mv elasticsearch-head-master /usr/local/elasticsearch-headcd /usr/local/elasticsearch-head # 安装依赖 npm install修改配置允许外部访问vim Gruntfile.js找到connect: { server: { options修改hostname为0.0.0.0connect: { server: { options: { hostname: 0.0.0.0, port: 9100, base: ., keepalive: true } } }# 后台启动 head端口 9100 nohup npm run start 访问验证浏览器访问http://服务器IP:9100在页面输入 ES 地址http://服务器IP:9200连接集群正确输出如下图注意事项防火墙 / 安全组需要放行 9100 端口ES 必须开启 CORS 跨域前面配置中已设置否则 head 无法连接4.2 Kibana 7.17.18 部署前提Elasticsearch 已经正常启动Kibana 版本必须和 ES 版本保持一致。4.2.1 下载解压cd /usr/local/src wget https://artifacts.elastic.co/downloads/kibana/kibana-7.17.18-linux-x86_64.tar.gztar -zxvf kibana-7.17.18-linux-x86_64.tar.gz mv kibana-7.17.18-linux-x86_64 /usr/local/kibana chown -R es:es /usr/local/kibana4.2.2 修改 kibana.yml 配置vim /usr/local/kibana/config/kibana.ymlserver.host: 0.0.0.0 server.name: my-kibana elasticsearch.hosts: [http://127.0.0.1:9200] i18n.locale: zh-CN参数说明server.host: 0.0.0.0允许外部浏览器访问 Kibanaelasticsearch.hosts指定连接 Elasticsearch 服务地址i18n.locale: zh-CN开启中文界面4.2.3 启动与访问su es cd /usr/local/kibana/bin ./kibana浏览器访问http://服务器IP:5601进入页面后打开【开发工具】就可以执行 ES 的 DSL 语句。五、日志清洗实战说明本章节用来演示日志清洗操作可填入你的原始日志样本、配置、操作步骤、清洗之后效果。5.1 业务需求说明需求 2完整版脚本ESMySQL 双写项目正式使用现在要做一个 web 日志收集项目现在已经将 nginx 日志存储到 kafkakafka 主机为 192.168.140.130nginx 存放 topic 为 nginxlog。系统搭建好了 elasticsearchelasticsearch 主机为 192.168.140.129索引为 nginxlog‑clean。现在需要写一个消费者脚本消费 kafka 日志将 nginx 原始日志清洗后存放到 elasticsearch 和 mysql 中。要求消费 nginx 原始日志原始日志格式为log_format main $remote_addr - $remote_user [$time_local] $request $status $body_bytes_sent $http_referer $http_user_agent $http_x_forwarded_for;提取出 ip、时间、请求接口、状态码、流量字段将时间格式转化为 2026‑05‑28 10:21:00提取出的字段以及转化后的时间存入 es 和 mysqles 如果没有相应索引就创建并按照字段的创建 mapping存在就直接存mysql 数据库和表没有就创建有就直接存配置文件单独一个5.2 部署实施5.2.1安装 Python 依赖库pip3 install kafka-python elasticsearch pymysql python-dateutil configparser5.2.2部署文件准备编写 config.ini 配置文件填写 Kafka、Elasticsearch、MySQL 连接信息。部署完整版 Python 消费脚本。5.2.3前台调试启动开发调试观察控制台输出cd /root python3 kafka_nginx_consumer.py # Ctrl C 终止前台运行程序5.2.4systemd 后台服务部署生产环境常驻# 重载 systemd 配置 systemctl daemon-reload 启动消费者服务 systemctl start nginx-kafka-consumer 设置开机自启 systemctl enable nginx-kafka-consumer 查看服务运行状态 systemctl status nginx-kafka-consumer 实时查看脚本控制台输出日志 journalctl -u nginx-kafka-consumer -f5.3 全链路验证命令1. 查看 kafka topic 消息总条数 cd /usr/local/kafka/bin ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.140.130:9092 --topic nginxlog 2. Kafka 控制台实时消费验证 Filebeat 消息是否正常推送 ./kafka-console-consumer.sh --bootstrap-server 192.168.140.130:9092 --topic nginxlog 3. 模拟产生 Nginx 新访问日志生成测试数据 curl http://192.168.140.129/ curl http://192.168.140.129/test/demo 4. 查询 ES 清洗后文档总数量 curl http://192.168.140.129:9200/nginxlog-clean/_count?pretty 5. 查看 ES 样例清洗完成数据 curl http://192.168.140.129:9200/nginxlog-clean/_search?pretty 6. MySQL 统计清洗日志行数 mysql -uroot -p -e use nginx_log_db;select count(*) from nginxlog_clean; 7. MySQL 查看 10 条样例数据 mysql -uroot -p -e use nginx_log_db;select ip,log_time,request_api,status from nginxlog_clean limit 10;5.4 历史存量日志消费说明Kafka 保存原始日志消息删除 ES 索引、truncate MySQL 表不会删除 Kafka 内部原始日志只是删除下游清洗后的结果可以重新消费清洗。修改config.ini配置使用全新从未使用过的 group_idearliest才会读取 topic 全部历史消息。测试环境可清空旧清洗结果启动脚本终端不再打印消费日志代表存量日志消费完成。3存量消费完毕停止脚本改回业务原有group_id切换latest模式只消费后续新增日志。5.5 清洗后结构化结果# 清洗完成后的文档示例 { _index: nginxlog-clean, _source: { ip: 192.168.10.20, log_time: 2026‑08‑17 21:30:45, remote_user: -, request_api: GET /index.html HTTP/1.1, status: 200, body_bytes_sent: 12580, http_referer: -, http_user_agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/118.0.0.0, http_x_forwarded_for: - } }5.6 效果验证在 Kibana 开发工具执行 DSL查看清洗完成的数据。六、常见问题排查can not run elasticsearch as rootES 禁止 root 启动切换 es 普通用户执行启动命令。启动报 vm.max_map_count 相关报错检查 /etc/sysctl.conf 配置执行 sysctl -p 使内核参数生效。外部访问 9200、5601 端口不通确认配置 network.host:0.0.0.0检查防火墙云服务器需要配置安全组放行端口。Kibana 无法连接 Elasticsearch确认 ES 服务正常运行核对 ES 与 Kibana 版本完全一致确认 elasticsearch.hosts 地址书写正确。IK 分词器不生效分词器版本必须和 ES 一致确认解压到 plugins/ik 目录修改完成后重启 Elasticsearch。七、总结ELK 三大组件分工明确Elasticsearch 负责存储检索Logstash 负责采集清洗Kibana 负责可视化展示。ES 依靠倒排索引实现高速检索分片副本解决存储与高可用。测试环境可以直接使用 ElasticsearchKibana 组合省略 Logstash 简化部署生产环境建议引入采集组件。ES 部署重点系统资源与内核调优、禁止 root 启动、组件版本严格对齐IK 分词器解决中文分词场景。本文为单机测试环境生产集群需要做 JVM 调优、索引生命周期管理等优化。