Flume cluster deployment
Install Flume and Basic Environment
Download and extract Flume
cd /export/software
wget -P /export/software/ https://repo.huaweicloud.com/apache/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz
sudo tar -zxvf /export/software/apache-flume-1.9.0-bin.tar.gz -C /export/servers/
cd /export/servers
sudo mv /export/servers/apache-flume-1.9.0-bin /export/servers/flumeConfigure environment variables
edit /etc/profilefile
sudo vim /etc/profile Add the following:
#Flume环境变量
export FLUME_HOME=/export/servers/flume
export PATH=$PATH:$FLUME_HOME/binMake configuration effective
source /etc/profileconfiguredflume-env.sh
sudo chown -R hadooper:hadooper /export/servers/flume/
cd $FLUME_HOME/conf
cp flume-env.sh.template flume-env.sh
sudo vi flume-env.shAdd the following:
export JAVA_HOME=/export/servers/jdkdistribution profile
scp -r /export/servers/flume hadoop02:/tmp/
scp /etc/profile hadoop02:/etc
scp -r /export/servers/flume hadoop03:/tmp/
scp -r /etc/profile hadoop03:/etcEnter hadoop02
sudo cp -r /tmp/flume /export/servers
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
rm -r /tmp/flumeEnter hadoop03
sudo cp -r /tmp/flume /export/servers
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
rm -r /tmp/flumeAfter the distribution is complete, enter two nodes to take effect for the configuration
source /etc/profileverify the installation
Executed on each node flume-ng version。If version information is displayed correctly, the basic installation is successful
Configure failover mode cluster nodes
Choose between failover mode and multi-layer data flow mode to configure!
Failover mode (recommended configuration) hadoop01: agent-node, responsible for collecting logs. hadoop02: collector1-node, receives data, high priority. hadoop03: collector2-node, receive data, alternate priority.
Failover mode configuration hadoop01 (agent-node)
cd /$FLUME_HOME/conf/create a profile
touch agent-failover.conf
vi /$FLUME_HOME/conf/agent-failover.confThe content is as follows:
# 定义组件名称
agent1.sources = r1
agent1.channels = c1
agent1.sinks = k1 k2
agent1.sinkgroups = g1
# 配置 Source：监控日志文件
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /export/servers/flume/weblog.log
agent1.sources.r1.channels = c1
# 配置 Channel：使用内存通道
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 10000
agent1.channels.c1.transactionCapacity = 10000
# 配置第一个 Sink（指向主 Collector）
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = hadoop02
agent1.sinks.k1.port = 52020
agent1.sinks.k1.channel = c1
# 配置第二个 Sink（指向备用 Collector）
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = hadoop03
agent1.sinks.k2.port = 52021
agent1.sinks.k2.channel = c1
# 配置 Sink Group 和故障转移策略
agent1.sinkgroups.g1.sinks = k1 k2
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 5
agent1.sinkgroups.g1.processor.maxpenalty = 10000Failover mode configuration hadoop02, hadoop03
Create configuration files separately on each Collector node
cd $FLUME_HOME/conf/
touch collector-hdfs.conf
vi /$FLUME_HOME/conf/collector-hdfs.confAdd the following: Failover Mode Hadoop02:
# 定义组件
collector1.sources = r1
collector1.channels = c1
collector1.sinks = k1
# 配置 Source：监听 Avro 端口
collector1.sources.r1.type = avro
collector1.sources.r1.bind = 0.0.0.0
collector1.sources.r1.port = 52020
collector1.sources.r1.channels = c1
# 配置 Channel
collector1.channels.c1.type = memory
collector1.channels.c1.capacity = 10000
collector1.channels.c1.transactionCapacity = 10000
# 配置 Sink：写入 HDFS
collector1.sinks.k1.type = hdfs
collector1.sinks.k1.channel = c1
collector1.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
collector1.sinks.k1.hdfs.filePrefix = events-
collector1.sinks.k1.hdfs.fileType = DataStream
collector1.sinks.k1.hdfs.writeFormat = Text
collector1.sinks.k1.hdfs.rollInterval = 3600
collector1.sinks.k1.hdfs.rollSize = 0
collector1.sinks.k1.hdfs.rollCount = 0
collector1.sinks.k1.hdfs.useLocalTimeStamp = trueFailover Mode Hadoop03:
# 定义组件
collector2.sources = r1
collector2.channels = c1
collector2.sinks = k1
# 配置 Source：监听 Avro 端口
collector2.sources.r1.type = avro
collector2.sources.r1.bind = 0.0.0.0
collector2.sources.r1.port = 52021
collector2.sources.r1.channels = c1
# 配置 Channel
collector2.channels.c1.type = memory
collector2.channels.c1.capacity = 10000
collector2.channels.c1.transactionCapacity = 10000
# 配置 Sink：写入 HDFS
collector2.sinks.k1.type = hdfs
collector2.sinks.k1.channel = c1
collector2.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
collector2.sinks.k1.hdfs.filePrefix = events-
collector2.sinks.k1.hdfs.fileType = DataStream
collector2.sinks.k1.hdfs.writeFormat = Text
collector2.sinks.k1.hdfs.rollInterval = 3600
collector2.sinks.k1.hdfs.rollSize = 0
collector2.sinks.k1.hdfs.rollCount = 0
collector2.sinks.k1.hdfs.useLocalTimeStamp = trueStart failover mode clustering and testing
Note: This step is the start and test of failover mode! Failover Mode:
Failover mode launches the Collector service
Executed on hadoop02 and hadoop03: hadoop02:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n collector1 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &hadoop03:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n collector2 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &Failover mode launches Agent services
Executed on hadoop01:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n agent1 -c conf -f $FLUME_HOME/conf/agent-failover.conf -Dflume.root.logger=INFO,console &Failover Mode Test Data Flow
Create test log files on hadoop01
echo "Test message 1 - $(date)" >> /export/servers/flume/weblog.log
echo "Test message 2 - $(date)" >> /export/servers/flume/weblog.logAgent log: Should show that events were collected and sent Collector1 (hadoop02) log: Should show that data was received and written to HDFS Collector2 (hadoop03) log: As a backup, data should not be received for the time being Check whether a data file is generated on HDFS
hdfs dfs -ls hdfs://mycluster/flume/events/
echo "=== 最新数据内容 ==="
hdfs dfs -cat hdfs://mycluster/flume/events/*/*.tmp | tail -10Failover Mode Test failover functionality
Stop Flume Collector on hadoop02
ps aux | grep flume | grep collector1
kill [进程ID]Create test log files on hadoop01
echo "Test message 1 - $(date)" >> /export/servers/flume/weblog.log
echo "Test message 2 - $(date)" >> /export/servers/flume/weblog.logObserve whether the Agent log displays switching to k2 (hadoop03) Confirm that hadoop03 starts receiving data and writing it to HDFS
Configure multi-layered data flow model cluster nodes
Choose between failover mode and multi-layer data flow mode to configure!
Multi-layered data flow model hadoop01: agent1-node, responsible for collecting logs. hadoop02: agent2-node, collector1-node, responsible for collecting logs, receiving data from nodes 1 and 2, hadoop03: agent3-node, collector2-node, responsible for collecting logs of this node and receiving data of this node
Multi-layer data flow mode configuration hadoop02, hadoop03
Multi-layered data flow model hadoop02:
in/export/servers/flume/confCreate one under the pathcollector-hdfs.conffile
cd /export/servers/flume/conf
touch collector-hdfs.conf
vi collector-hdfs.confAnd add the following:
# 定义组件
collector1.sources = r1
collector1.channels = c1
collector1.sinks = k1
# 配置 Source：监听 Avro 端口
collector1.sources.r1.type = avro
collector1.sources.r1.bind = 0.0.0.0
collector1.sources.r1.port = 52020
collector1.sources.r1.channels = c1
# 配置 Channel
collector1.channels.c1.type = memory
collector1.channels.c1.capacity = 10000
collector1.channels.c1.transactionCapacity = 10000
# 配置 Sink：写入 HDFS
collector1.sinks.k1.type = hdfs
collector1.sinks.k1.channel = c1
collector1.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
collector1.sinks.k1.hdfs.filePrefix = events-
collector1.sinks.k1.hdfs.fileType = DataStream
collector1.sinks.k1.hdfs.writeFormat = Text
collector1.sinks.k1.hdfs.rollInterval = 3600
collector1.sinks.k1.hdfs.rollSize = 0
collector1.sinks.k1.hdfs.rollCount = 0
collector1.sinks.k1.hdfs.useLocalTimeStamp = trueMulti-layered data flow model hadoop03:
in/export/servers/flume/confCreate one under the pathcollector-hdfs.conffile
cd /export/servers/flume/conf
touch collector-hdfs.conf
vi collector-hdfs.confAnd add the following:
# 定义组件
collector2.sources = r1
collector2.sinks = k1
collector2.channels = c1
# 配置 Avro Source - 使用不同的端口（如 52021）
collector2.sources.r1.type = avro
collector2.sources.r1.bind = 0.0.0.0
collector2.sources.r1.port = 52021
collector2.sources.r1.channels = c1
# 添加时间戳拦截器
collector2.sources.r1.interceptors = i1
collector2.sources.r1.interceptors.i1.type = timestamp
collector2.sources.r1.interceptors.i1.preserveExisting = false
# 配置 HDFS Sink - 使用不同的 HDFS 路径
collector2.sinks.k1.type = hdfs
collector2.sinks.k1.channel = c1
collector2.sinks.k1.hdfs.path = hdfs://mycluster/flume/events/%Y-%m-%d/
collector2.sinks.k1.hdfs.filePrefix = events-
collector2.sinks.k1.hdfs.round = true
collector2.sinks.k1.hdfs.roundValue = 10
collector2.sinks.k1.hdfs.roundUnit = minute
collector2.sinks.k1.hdfs.rollInterval = 3600
collector2.sinks.k1.hdfs.rollSize = 0
collector2.sinks.k1.hdfs.rollCount = 0
collector2.sinks.k1.hdfs.batchSize = 1000
collector2.sinks.k1.hdfs.fileType = DataStream
# 配置 Memory Channel
collector2.channels.c1.type = memory
collector2.channels.c1.capacity = 1000
collector2.channels.c1.transactionCapacity = 1000The multi-layered data flow model creates separate profiles for each Agent node
Multi-layered data flow model hadoop01:
enter/export/servers/flume/conf/directory
cd /export/servers/flume/conf/create a new profile
touch agent-node1.conf
vim agent-node1.confAnd add the following:
# 定义组件名称
agent1.sources = r1
agent1.channels = c1
agent1.sinks = k1
# 配置 Source：监控日志文件
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop01.log
agent1.sources.r1.channels = c1
# 配置 Channel：使用内存通道
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 10000
agent1.channels.c1.transactionCapacity = 10000
# 配置第一个 Sink（指向主 Collector）
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = hadoop02
agent1.sinks.k1.port = 52020
agent1.sinks.k1.channel = c1Enter hadoop02 and hadoop03 respectively to modify the configuration files:
Multi-layered data flow model hadoop02:
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
cd /export/servers/flume/conf
vi agent-node2.confThe document is as follows:
# 定义组件名称
agent2.sources = r1
agent2.channels = c1
agent2.sinks = k1
# 配置 Source：监控日志文件
agent2.sources.r1.type = exec
agent2.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop02.log
agent2.sources.r1.channels = c1
# 配置 Channel：使用内存通道
agent2.channels.c1.type = memory
agent2.channels.c1.capacity = 10000
agent2.channels.c1.transactionCapacity = 10000
# 配置第一个 Sink（指向主 Collector）
agent2.sinks.k1.type = avro
agent2.sinks.k1.hostname = hadoop02
agent2.sinks.k1.port = 52020
agent2.sinks.k1.channel = c1Multi-layered data flow model hadoop03:
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
cd /export/servers/flume/conf
vi agent-node3.confThe document is as follows:
# 定义组件名称
agent3.sources = r1
agent3.channels = c1
agent3.sinks = k1
# 配置 Source：监控日志文件
agent3.sources.r1.type = exec
agent3.sources.r1.command = tail -F /export/servers/flume/weblog-hadoop03.log
agent3.sources.r1.channels = c1
# 配置 Channel：使用内存通道
agent3.channels.c1.type = memory
agent3.channels.c1.capacity = 10000
agent3.channels.c1.transactionCapacity = 10000
# 配置第一个 Sink（指向主 Collector）
agent3.sinks.k1.type = avro
agent3.sinks.k1.hostname = hadoop03
agent3.sinks.k1.port = 52021
agent3.sinks.k1.channel = c1Start cluster and test
This step is to start and test the multi-layered data flow model!
Multi-layered data flow model:
Multi-layered data flow model launches Collector service
Executed on hadoop02 and hadoop03: Multi-layered data flow model hadoop02:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n collector1 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &Multi-layered data flow model hadoop03:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n collector2 -c conf -f $FLUME_HOME/conf/collector-hdfs.conf -Dflume.root.logger=INFO,console &Launching Agent services in multi-layered data flow model
Multi-layered data flow model hadoop01:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n agent1 -c conf -f $FLUME_HOME/conf/agent-node1.conf -Dflume.root.logger=INFO,console &Multi-layered data flow model hadoop02:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n agent2 -c conf -f $FLUME_HOME/conf/agent-node2.conf -Dflume.root.logger=INFO,console &Multi-layered data flow model hadoop03:
cd $FLUME_HOME
$FLUME_HOME/bin/flume-ng agent -n agent3 -c conf -f $FLUME_HOME/conf/agent-node3.conf -Dflume.root.logger=INFO,console &Multi-layer data flow pattern tests data flow
Test on hadoop01 (Send to hadoop02)
echo "Agent1 to Collector1 测试 - $(date)" >> /export/servers/flume/weblog-hadoop01.logTest on hadoop02 (Send to hadoop02)
echo " Agent2 to Collector1 测试 - $(date) " >> /export/servers/flume/weblog-hadoop02.logTest on hadoop03 (Send to local hadoop03)
echo " Agent3 to Collector2 测试 - $(date) " >> /export/servers/flume/weblog-hadoop03.logVerify HDFS data writes
hdfs dfs -ls hdfs://mycluster/flume/events/
echo "=== 最新数据内容 ==="
hdfs dfs -cat hdfs://mycluster/flume/events/*/*.tmp | tail -10
If you enjoyed this, leave a comment~