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/flume

Configure environment variables

Editing /etc/profile files

sudo vim /etc/profile 

Add the following:

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

Make configuration effective

source /etc/profile

Configure flume-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.sh

Add the following:

export JAVA_HOME=/export/servers/jdk

distribution 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:/etc

Enter hadoop02

sudo cp -r /tmp/flume /export/servers
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
rm -r /tmp/flume

Enter hadoop03

sudo cp -r /tmp/flume /export/servers
cd /export/servers
sudo chown -R hadooper:hadooper /export/servers/flume/
rm -r /tmp/flume

After the distribution is complete, enter two nodes to take effect for the configuration

source /etc/profile

verify the installation

flume-ng version is executed on each node. 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.conf

The 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 = 10000

Failover 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.conf

Add 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 = true

Failover 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 = true

Start 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.log

Agent 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 -10

Failover 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.log

Observe 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: Create a collector-hdfs.conf file under the /export/servers/flume/conf path

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

And 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 = true

Multi-layered data flow model hadoop03: Create a collector-hdfs.conf file under the /export/servers/flume/conf path

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

And 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 = 1000

Multi-tiered data flow patterns create separate profiles for each Agent node

Multi-layer data stream mode hadoop01: Enter /export/servers/flume/conf/ catalog

cd /export/servers/flume/conf/

create a new profile

touch agent-node1.conf
vim agent-node1.conf

And 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 = c1

Enter 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.conf

The 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 = c1

Multi-layered data flow model hadoop03:

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

The 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 = c1

Start 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 the 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-layer data stream mode 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:52020)

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

Test on hadoop02 (Send to hadoop02:52020)

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

Test on hadoop03 (send to local hadoop03:52021)

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

Verify HDFS data writes

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