部署Flink、Kafka、Elasticsearch和Kibana的集群

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

目录

前言

一、前期构思

二、搭建步骤

1.搭建虚拟机集群（CentOS 7）

2.安装ES7.17.21

​编辑​编辑3.安装Flink1.19.3

4.kafka配置

5.配置Kibana

6.配置一键启动脚本

总结

前言

由于本人是网安大三牲，学校要求设计毕设，本人升本之前又是大数据专业的本科专业知识学的一知半解、于是乎决定搭建一个基于大数据流处理流量异常监测平台

一、前期构思

面对日益复杂的网络攻击与海量数据流，构建现代化的网络安全防御体系，必须满足三个核心要求：实时响应、高效处理与弹性扩展。为此，我选择了以 Apache Flink 作为实时计算引擎、Apache Kafka 作为数据采集与缓冲通道、Elasticsearch 作为存储与检索引擎，并结合 Kibana 实现可视化分析的整套技术栈。

在具体的版本选择上，原本计划采用较新的 Elasticsearch 8.13.1 与 Flink 1.19.3 组合。但在实际集成过程中发现，Flink 官方提供的 Elasticsearch 连接器（Connector）对其新版客户端的支持尚不完善，未能找到完全兼容这两个版本的稳定连接方案。为确保整套流水线的稳定与可维护性，最终决定将 Elasticsearch 版本适度回调至一个与 Flink 连接器生态兼容性更好的稳定版本。这一调整体现了在实际工程中，技术选型不仅需要考虑组件的前沿性，更要优先确保整个系统链条的可靠衔接与稳健运行。

二、搭建步骤

1.搭建虚拟机集群（CentOS 7）

由于我的电脑配置较差于是我选择配属三台虚拟机作为集群节点配置如下：

（主机1与主机2配置相同）

主机3配置如下：

为了方便后续查找与配置我选择在home目录下新建一个文件夹为bigdata（三台主机目录配相同方便后续集群分发）

2.安装ES7.17.21

4.kafka配置

跟上述一样下载解压安装

修改/conf/server.properties文件

# ==============================================================================
# 1. 基础配置 (每个节点必须唯一)
# ==============================================================================
# 节点 ID，集群内唯一。101 设为 0，102 设为 1，103 设为 2
broker.id=0

# ==============================================================================
# 2. 网络与监听配置 (根据节点 IP 修改)
# ==============================================================================
# 监听地址
listeners=PLAINTEXT://192.168.10.101:9092
# 告诉客户端如何连接到当前 Broker
advertised.listeners=PLAINTEXT://192.168.10.101:9092

# 处理网络请求的线程数
num.network.threads=3
# 处理磁盘 I/O 的线程数
num.io.threads=8

# ==============================================================================
# 3. 日志与存储配置
# ==============================================================================
# 日志文件存储路径（确保该目录存在且有写入权限）
log.dirs=/tmp/kafka-logs

# 每个 Topic 默认的分区数
num.partitions=3
# 每个分区的副本因子，建议设为 2（保证高可用）
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
transaction.state.log.min.isr=2

# 日志保留策略（保留 168 小时 / 7 天）
log.retention.hours=168
# 单个日志段文件的最大大小
log.segment.bytes=1073741824

# ==============================================================================
# 4. Zookeeper 集群配置 (三台机器写死一致)
# ==============================================================================
zookeeper.connect=192.168.10.101:2181,192.168.10.102:2181,192.168.10.103:2181
# Zookeeper 连接超时时间
zookeeper.connection.timeout.ms=18000

5.配置Kibana

下载重解压后修改config/kibana.yml文件

# ==============================================================================
# 基础服务配置
# ==============================================================================
# 默认端口 5601
server.port: 5601

# 允许外部浏览器访问（设为 0.0.0.0 会监听所有网卡）
server.host: "0.0.0.0"

# 服务的公共访问地址（用于生成分享链接）
server.publicBaseUrl: "http://192.168.10.101:5601"

# ==============================================================================
# Elasticsearch 连接配置 (三节点集群)
# ==============================================================================
# 这里填入你的 ES 集群节点，Kibana 会自动进行负载均衡
elasticsearch.hosts:
– "http://192.168.10.101:9200"
– "http://192.168.10.102:9200"
– "http://192.168.10.103:9200"

# 如果你在 ES 中开启了安全验证，请取消下面两行的注释并填写账号密码
# elasticsearch.username: "kibana_system"
# elasticsearch.password: "your_password"

# ==============================================================================
# 界面语言设置
# ==============================================================================
# 强烈建议设置为中文，方便监控项配置
i18n.locale: "zh-CN"

6.配置一键启动脚本

为方便启动，使用ai配置了一个以前启动脚本

#!/bin/bash

# — 基础配置 —
NODES=("192.168.10.101" "192.168.10.102" "192.168.10.103")
KIBANA_NODE="192.168.10.101"
# 运行 ES 的普通用户名（必须非 root）
ES_USER="elastic"
# ES 7 的安装路径
ES_PATH="/home/bigdata/elasticsearch-7"
# Kibana 的安装路径
KIBANA_PATH="/home/bigdata/kibana-7.17"

echo "====================================================="
echo " 异常分析组件集群一键启动脚本 (ES 7.x 版) "
echo "====================================================="

# 1. ZooKeeper & 2. Kafka (保持你原有的逻辑)
echo ">>> [1/5] 启动 ZooKeeper & Kafka 集群…"
for HOST in "${NODES[@]}"; do
ssh $HOST "source /etc/profile; /home/bigdata/kafka/bin/zookeeper-server-start.sh -daemon /home/bigdata/kafka/config/zookeeper.properties"
ssh $HOST "source /etc/profile; /home/bigdata/kafka/bin/kafka-server-start.sh -daemon /home/bigdata/kafka/config/server.properties"
echo " – $HOST: ZK & Kafka 已启动"
done

# 3. Elasticsearch (降级后的核心改动)
echo ">>> [3/5] 启动 Elasticsearch 7 集群…"
for HOST in "${NODES[@]}"; do
# 使用 su 切换到普通用户执行启动命令，-d 为后台运行
ssh $HOST "su – $ES_USER -c 'nohup $ES_PATH/bin/elasticsearch -d > /dev/null 2>&1 &'"
echo " – $HOST: ES 指令已以 $ES_USER 身份下发"
done

echo "等待 ES 集群初始化 (40s)…"
sleep 40
echo "====================================================="
echo " 异常分析组件集群一键启动脚本 (ES 7.x 版) "
echo "====================================================="

# 1. ZooKeeper & 2. Kafka
echo ">>> [1/5] 启动 ZooKeeper & Kafka 集群…"
for HOST in "${NODES[@]}"; do
ssh $HOST "source /etc/profile; /home/bigdata/kafka/bin/zookeeper-server-start.sh -daemon /home/bigdata/kafka/config/zookeeper.properties"
ssh $HOST "source /etc/profile; /home/bigdata/kafka/bin/kafka-server-start.sh -daemon /home/bigdata/kafka/config/server.properties"
echo " – $HOST: ZK & Kafka 已启动"
done

# 3. Elasticsearch (降级后的核心改动)
echo ">>> [3/5] 启动 Elasticsearch 7 集群…"
for HOST in "${NODES[@]}"; do
# 使用 su 切换到普通用户执行启动命令，-d 为后台运行
ssh $HOST "su – $ES_USER -c 'nohup $ES_PATH/bin/elasticsearch -d > /dev/null 2>&1 &'"
echo " – $HOST: ES 指令已以 $ES_USER 身份下发"

赋予权限  chmod +x cluster-start.sh 

启动后检查服务

登录192.168.10.101：8081，192.168.10.101：5601

192.168.10.101：8081

192.168.10.101：5601

总结

至此部署成功，一个大致框架构成，后续想的是是否可以加入机器学习监测，目前仅使用过py脚本生成相对格式的流量访问但是没有真实数据，下一步打算使用权威发布的数据集进行实验查看是否可行，但是更希望的是可以采集真实流量。其实也不确定可行性高不高