SuifengFlying

公告

欢迎来到我的博客，这里记录生活与技术。

了解更多

1927 字

10 分钟

Flume集群部署

2025-11-24

BigData

/

Flume

/

Deployment

Flume集群部署#

安装Flume与基础环境#

下载、解压Flume#

1
cd /export/software
2
wget -P /export/software/ https://repo.huaweicloud.com/apache/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz
3
sudo tar -zxvf /export/software/apache-flume-1.9.0-bin.tar.gz -C /export/servers/
4
cd /export/servers
5
sudo mv /export/servers/apache-flume-1.9.0-bin /export/servers/flume

配置环境变量#

编辑 /etc/profile文件

1
sudo vim /etc/profile

添加以下内容：

1
#Flume环境变量
2
export FLUME_HOME=/export/servers/flume
3
export PATH=$PATH:$FLUME_HOME/bin

使配置生效

1
source /etc/profile

配置flume-env.sh#

1
sudo chown -R hadooper:hadooper /export/servers/flume/
2
cd $FLUME_HOME/conf
3
cp flume-env.sh.template flume-env.sh
4
sudo vi flume-env.sh

添加以下内容：

1
export JAVA_HOME=/export/servers/jdk

分发配置文件#

1
scp -r /export/servers/flume hadoop02:/tmp/
2
scp /etc/profile hadoop02:/etc
3
scp -r /export/servers/flume hadoop03:/tmp/
4
scp -r /etc/profile hadoop03:/etc

进入hadoop02:#

1
sudo cp -r /tmp/flume /export/servers
2
cd /export/servers
3
sudo chown -R hadooper:hadooper /export/servers/flume/
4
rm -r /tmp/flume

进入hadoop03:#

1
sudo cp -r /tmp/flume /export/servers
2
cd /export/servers
3
sudo chown -R hadooper:hadooper /export/servers/flume/
4
rm -r /tmp/flume

分发完成后进入两个节点中使配置生效#

1
source /etc/profile

验证安装#

每个节点上执行 flume-ng version。如果正确显示版本信息，则基础安装成功

配置故障转移模式集群节点#

故障转移模式和多层数据流模式二选一进行配置即可！

故障转移模式（建议配置） hadoop01：agent-node，负责采集日志。 hadoop02：collector1-node，接收数据，高优先级。 hadoop03：collector2-node，接收数据，备用优先级。

故障转移模式配置 hadoop01 (agent-node)#

1
cd /$FLUME_HOME/conf/

创建配置文件

1
touch agent-failover.conf
2
vi /$FLUME_HOME/conf/agent-failover.conf

内容如下：

1
# 定义组件名称
2
agent1.sources = r1
3
agent1.channels = c1
4
agent1.sinks = k1 k2
5
agent1.sinkgroups = g1
6

7
# 配置 Source：监控日志文件
8
agent1.sources.r1.type = exec
9
agent1.sources.r1.command = tail -F /export/servers/flume/weblog.log
10
agent1.sources.r1.channels = c1
11

12
# 配置 Channel：使用内存通道
13
agent1.channels.c1.type = memory
14
agent1.channels.c1.capacity = 10000
15
agent1.channels.c1.transactionCapacity = 10000
16

17
# 配置第一个 Sink（指向主 Collector）
18
agent1.sinks.k1.type = avro
19
agent1.sinks.k1.hostname = hadoop02
20
agent1.sinks.k1.port = 52020
21
agent1.sinks.k1.channel = c1
22

23
# 配置第二个 Sink（指向备用 Collector）
24
agent1.sinks.k2.type = avro
25
agent1.sinks.k2.hostname = hadoop03
26
agent1.sinks.k2.port = 52021
27
agent1.sinks.k2.channel = c1
28

29
# 配置 Sink Group 和故障转移策略
30
agent1.sinkgroups.g1.sinks = k1 k2
31
agent1.sinkgroups.g1.processor.type = failover
32
agent1.sinkgroups.g1.processor.priority.k1 = 10
33
agent1.sinkgroups.g1.processor.priority.k2 = 5
34
agent1.sinkgroups.g1.processor.maxpenalty = 10000

故障转移模式配置 hadoop02、hadoop03#

分别在每个 Collector 节点上创建配置文件

1
cd $FLUME_HOME/conf/
2
touch collector-hdfs.conf
3
vi /$FLUME_HOME/conf/collector-hdfs.conf

添加内容如下：故障转移模式hadoop02:

1
# 定义组件
2
collector1.sources = r1
3
collector1.channels = c1
4
collector1.sinks = k1
5

6
# 配置 Source：监听 Avro 端口
7
collector1.sources.r1.type = avro
8
collector1.sources.r1.bind = 0.0.0.0
9
collector1.sources.r1.port = 52020
10
collector1.sources.r1.channels = c1
11

12
# 配置 Channel
13
collector1.channels.c1.type = memory
14
collector1.channels.c1.capacity = 10000
15
collector1.channels.c1.transactionCapacity = 10000
16

17
# 配置 Sink：写入 HDFS
18
collector1.sinks.k1.type = hdfs
19
collector1.sinks.k1.channel = c1
20
collector1.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
21
collector1.sinks.k1.hdfs.filePrefix = events-
22
collector1.sinks.k1.hdfs.fileType = DataStream
23
collector1.sinks.k1.hdfs.writeFormat = Text
24
collector1.sinks.k1.hdfs.rollInterval = 3600
25
collector1.sinks.k1.hdfs.rollSize = 0
26
collector1.sinks.k1.hdfs.rollCount = 0
27
collector1.sinks.k1.hdfs.useLocalTimeStamp = true

故障转移模式hadoop03:

1
# 定义组件
2
collector2.sources = r1
3
collector2.channels = c1
4
collector2.sinks = k1
5

6
# 配置 Source：监听 Avro 端口
7
collector2.sources.r1.type = avro
8
collector2.sources.r1.bind = 0.0.0.0
9
collector2.sources.r1.port = 52021
10
collector2.sources.r1.channels = c1
11

12
# 配置 Channel
13
collector2.channels.c1.type = memory
14
collector2.channels.c1.capacity = 10000
15
collector2.channels.c1.transactionCapacity = 10000
16

17
# 配置 Sink：写入 HDFS
18
collector2.sinks.k1.type = hdfs
19
collector2.sinks.k1.channel = c1
20
collector2.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
21
collector2.sinks.k1.hdfs.filePrefix = events-
22
collector2.sinks.k1.hdfs.fileType = DataStream
23
collector2.sinks.k1.hdfs.writeFormat = Text
24
collector2.sinks.k1.hdfs.rollInterval = 3600
25
collector2.sinks.k1.hdfs.rollSize = 0
26
collector2.sinks.k1.hdfs.rollCount = 0
27
collector2.sinks.k1.hdfs.useLocalTimeStamp = true

启动故障转移模式集群与测试#

注：此步骤为故障转移模式的启动与测试！故障转移模式:

故障转移模式启动 Collector 服务#

在hadoop02和hadoop03上执行： hadoop02:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n collector1 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &

hadoop03:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n collector2 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &

故障转移模式启动 Agent 服务#

在hadoop01上执行：

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n agent1 -c conf -f $FLUME_HOME/conf/agent-failover.conf -Dflume.root.logger=INFO,console &

故障转移模式测试数据流#

在 hadoop01 上创建测试日志文件

1
echo "Test message 1 - $(date)" >> /export/servers/flume/weblog.log
2

3
echo "Test message 2 - $(date)" >> /export/servers/flume/weblog.log

Agent 日志：应该显示事件被采集和发送 Collector1 (hadoop02) 日志：应该显示接收到数据并写入 HDFS Collector2 (hadoop03) 日志：作为备用，暂时不应接收数据检查 HDFS 上是否生成了数据文件

1
hdfs dfs -ls hdfs://mycluster/flume/events/
2
echo "=== 最新数据内容 ==="
3
hdfs dfs -cat hdfs://mycluster/flume/events/*/*.tmp | tail -10

故障转移模式测试故障转移功能#

在 hadoop02 上停止 Flume Collector

1
ps aux | grep flume | grep collector1
2
kill [进程ID]

在 hadoop01 上创建测试日志文件

1
echo "Test message 1 - $(date)" >> /export/servers/flume/weblog.log
2

3
echo "Test message 2 - $(date)" >> /export/servers/flume/weblog.log

观察 Agent 日志是否显示切换到 k2 (hadoop03) 确认 hadoop03 开始接收数据并写入 HDFS

配置多层数据流模式集群节点#

故障转移模式和多层数据流模式二选一进行配置即可！

多层数据流模式 hadoop01：agent1-node，负责采集日志。 hadoop02：agent2-node, collector1-node，负责采集日志,接收1，2节点数据， hadoop03：agent3-node, collector2-node，负责采集本节点日志,接收本节点数据

多层数据流模式配置hadoop02、hadoop03#

多层数据流模式hadoop02: 在/export/servers/flume/conf路径下创建一个collector-hdfs.conf文件

1
cd /export/servers/flume/conf
2
touch collector-hdfs.conf
3
vi collector-hdfs.conf

并添加以下内容：

1
# 定义组件
2
collector1.sources = r1
3
collector1.channels = c1
4
collector1.sinks = k1
5

6
# 配置 Source：监听 Avro 端口
7
collector1.sources.r1.type = avro
8
collector1.sources.r1.bind = 0.0.0.0
9
collector1.sources.r1.port = 52020
10
collector1.sources.r1.channels = c1
11

12
# 配置 Channel
13
collector1.channels.c1.type = memory
14
collector1.channels.c1.capacity = 10000
15
collector1.channels.c1.transactionCapacity = 10000
16

17
# 配置 Sink：写入 HDFS
18
collector1.sinks.k1.type = hdfs
19
collector1.sinks.k1.channel = c1
20
collector1.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
21
collector1.sinks.k1.hdfs.filePrefix = events-
22
collector1.sinks.k1.hdfs.fileType = DataStream
23
collector1.sinks.k1.hdfs.writeFormat = Text
24
collector1.sinks.k1.hdfs.rollInterval = 3600
25
collector1.sinks.k1.hdfs.rollSize = 0
26
collector1.sinks.k1.hdfs.rollCount = 0
27
collector1.sinks.k1.hdfs.useLocalTimeStamp = true

多层数据流模式hadoop03: 在/export/servers/flume/conf路径下创建一个collector-hdfs.conf文件

1
cd /export/servers/flume/conf
2
touch collector-hdfs.conf
3
vi collector-hdfs.conf

并添加以下内容：

1
# 定义组件
2
collector2.sources = r1
3
collector2.sinks = k1
4
collector2.channels = c1
5

6
# 配置 Avro Source - 使用不同的端口（如 52021）
7
collector2.sources.r1.type = avro
8
collector2.sources.r1.bind = 0.0.0.0
9
collector2.sources.r1.port = 52021
10
collector2.sources.r1.channels = c1
11
# 添加时间戳拦截器
12
collector2.sources.r1.interceptors = i1
13
collector2.sources.r1.interceptors.i1.type = timestamp
14
collector2.sources.r1.interceptors.i1.preserveExisting = false
15

16
# 配置 HDFS Sink - 使用不同的 HDFS 路径
17
collector2.sinks.k1.type = hdfs
18
collector2.sinks.k1.channel = c1
19
collector2.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
20
collector2.sinks.k1.hdfs.filePrefix = events-
21
collector2.sinks.k1.hdfs.round = true
22
collector2.sinks.k1.hdfs.roundValue = 10
23
collector2.sinks.k1.hdfs.roundUnit = minute
24
collector2.sinks.k1.hdfs.rollInterval = 3600
25
collector2.sinks.k1.hdfs.rollSize = 0
26
collector2.sinks.k1.hdfs.rollCount = 0
27
collector2.sinks.k1.hdfs.batchSize = 1000
28
collector2.sinks.k1.hdfs.fileType = DataStream
29

30
# 配置 Memory Channel
31
collector2.channels.c1.type = memory
32
collector2.channels.c1.capacity = 1000
33
collector2.channels.c1.transactionCapacity = 1000

多层数据流模式为每个 Agent 节点创建单独的配置文件#

多层数据流模式hadoop01: 进入/export/servers/flume/conf/目录

1
cd /export/servers/flume/conf/

创建新配置文件

1
touch agent-node1.conf
2
vim agent-node1.conf

并添加以下内容:

1
# 定义组件名称
2
agent1.sources = r1
3
agent1.channels = c1
4
agent1.sinks = k1
5

6
# 配置 Source：监控日志文件
7
agent1.sources.r1.type = exec
8
agent1.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop01.log
9
agent1.sources.r1.channels = c1
10

11
# 配置 Channel：使用内存通道
12
agent1.channels.c1.type = memory
13
agent1.channels.c1.capacity = 10000
14
agent1.channels.c1.transactionCapacity = 10000
15

16
# 配置第一个 Sink（指向主 Collector）
17
agent1.sinks.k1.type = avro
18
agent1.sinks.k1.hostname = hadoop02
19
agent1.sinks.k1.port = 52020
20
agent1.sinks.k1.channel = c1

分别进入hadoop02、hadoop03修改配置文件：

多层数据流模式hadoop02:

1
cd /export/servers
2
sudo chown -R hadooper:hadooper /export/servers/flume/
3
cd /export/servers/flume/conf
4
vi agent-node2.conf

文件内容如下：

1
# 定义组件名称
2
agent2.sources = r1
3
agent2.channels = c1
4
agent2.sinks = k1
5

6
# 配置 Source：监控日志文件
7
agent2.sources.r1.type = exec
8
agent2.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop02.log
9
agent2.sources.r1.channels = c1
10

11
# 配置 Channel：使用内存通道
12
agent2.channels.c1.type = memory
13
agent2.channels.c1.capacity = 10000
14
agent2.channels.c1.transactionCapacity = 10000
15

16
# 配置第一个 Sink（指向主 Collector）
17
agent2.sinks.k1.type = avro
18
agent2.sinks.k1.hostname = hadoop02
19
agent2.sinks.k1.port = 52020
20
agent2.sinks.k1.channel = c1

多层数据流模式hadoop03:

1
cd /export/servers
2
sudo chown -R hadooper:hadooper /export/servers/flume/
3
cd /export/servers/flume/conf
4
vi agent-node3.conf

文件内容如下：

1
# 定义组件名称
2
agent3.sources = r1
3
agent3.channels = c1
4
agent3.sinks = k1
5

6
# 配置 Source：监控日志文件
7
agent3.sources.r1.type = exec
8
agent3.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop03.log
9
agent3.sources.r1.channels = c1
10

11
# 配置 Channel：使用内存通道
12
agent3.channels.c1.type = memory
13
agent3.channels.c1.capacity = 10000
14
agent3.channels.c1.transactionCapacity = 10000
15

16
# 配置第一个 Sink（指向主 Collector）
17
agent3.sinks.k1.type = avro
18
agent3.sinks.k1.hostname = hadoop03
19
agent3.sinks.k1.port = 52021
20
agent3.sinks.k1.channel = c1

启动集群与测试#

此步骤为多层数据流模式的启动与测试！

多层数据流模式:

多层数据流模式启动 Collector 服务#

在hadoop02和hadoop03上执行：多层数据流模式hadoop02:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n collector1 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &

多层数据流模式hadoop03:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n collector2 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &

多层数据流模式启动 Agent 服务：#

多层数据流模式hadoop01:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n agent1 -c conf -f $FLUME_HOME/conf/agent-node1.conf -Dflume.root.logger=INFO,console &

多层数据流模式hadoop02:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n agent2 -c conf -f $FLUME_HOME/conf/agent-node2.conf -Dflume.root.logger=INFO,console &

多层数据流模式hadoop03:

1
cd $FLUME_HOME
2
$FLUME_HOME/bin/flume-ng agent -n agent3 -c conf -f $FLUME_HOME/conf/agent-node3.conf -Dflume.root.logger=INFO,console &

多层数据流模式测试数据流#

在 hadoop01 上测试 (发送到 hadoop02:52020)#

1
echo "Agent1 to Collector1 测试 - $(date)" >> /export/servers/flume/weblog-hadoop01.log

在 hadoop02 上测试 (发送到 hadoop02:52020)#

1
echo " Agent2 to Collector1 测试 - $(date) " >> /export/servers/flume/weblog-hadoop02.log

在 hadoop03 上测试 (发送到本地 hadoop03:52021)#

1
echo " Agent3 to Collector2 测试 - $(date) " >> /export/servers/flume/weblog-hadoop03.log

验证 HDFS 数据写入#

1
hdfs dfs -ls hdfs://mycluster/flume/events/
2
echo "=== 最新数据内容 ==="
3
hdfs dfs -cat hdfs://mycluster/flume/events/*/*.tmp | tail -10

Flume集群部署

https://mizuki.mysqil.com/posts/flume部署/

作者

SuifengFlying

发布于

2025-11-24

许可协议

CC BY-NC-SA 4.0

Big Data Deployment

2 / 12

Flink部署 Hadoop版本选择

欢迎来到 SuifengFlying!

Flume集群部署#

安装Flume与基础环境#

下载、解压Flume#

配置环境变量#

配置flume-env.sh#

分发配置文件#

进入hadoop02:#

进入hadoop03:#

分发完成后进入两个节点中使配置生效#

验证安装#

配置故障转移模式集群节点#

故障转移模式配置 hadoop01 (agent-node)#

故障转移模式配置 hadoop02、hadoop03#

启动故障转移模式集群与测试#

故障转移模式启动 Collector 服务#

故障转移模式启动 Agent 服务#

故障转移模式测试数据流#

故障转移模式测试故障转移功能#

配置多层数据流模式集群节点#

多层数据流模式配置hadoop02、hadoop03#

多层数据流模式为每个 Agent 节点创建单独的配置文件#

启动集群与测试#

多层数据流模式启动 Collector 服务#

多层数据流模式启动 Agent 服务：#

多层数据流模式测试数据流#

在 hadoop01 上测试 (发送到 hadoop02:52020)#

在 hadoop02 上测试 (发送到 hadoop02:52020)#

在 hadoop03 上测试 (发送到本地 hadoop03:52021)#

验证 HDFS 数据写入#

Big Data Deployment

Related articles

欢迎来到 SuifengFlying!

Flume集群部署#

安装Flume与基础环境#

下载、解压Flume#

配置环境变量#

配置flume-env.sh#

分发配置文件#

进入hadoop02:#

进入hadoop03:#

分发完成后进入两个节点中使配置生效#

验证安装#

配置故障转移模式集群节点#

故障转移模式配置 hadoop01 (agent-node)#

故障转移模式配置 hadoop02、hadoop03#

启动故障转移模式集群与测试#

故障转移模式启动 Collector 服务#

故障转移模式启动 Agent 服务#

故障转移模式测试数据流#

故障转移模式测试故障转移功能#

配置多层数据流模式集群节点#

多层数据流模式配置hadoop02、hadoop03#

多层数据流模式为每个 Agent 节点创建单独的配置文件#

启动集群与测试#

多层数据流模式启动 Collector 服务#

多层数据流模式启动 Agent 服务：#

多层数据流模式测试数据流#

在 hadoop01 上测试​ (发送到 hadoop02:52020)#

在 hadoop02 上测试​ (发送到 hadoop02:52020)#

在 hadoop03 上测试​ (发送到本地 hadoop03:52021)#

验证 HDFS 数据写入#

Big Data Deployment

Related articles

在 hadoop01 上测试 (发送到 hadoop02:52020)#

在 hadoop02 上测试 (发送到 hadoop02:52020)#

在 hadoop03 上测试 (发送到本地 hadoop03:52021)#