Flume NG 学习笔记（二）安装部署以及实战案例

àì夳堔傛蜴生んèń 2022-05-16 01:37 214阅读 0赞

### 安装部署 Flume ###

1、Flume 的安装非常简单，只需要解压即可，当然，前提是已有 Hadoop 环境上传安装包到 数据源所在节点上

然后解压 tar -zxvf apache-flume-1.8.0-bin.tar.gz

然后进入 flume 的目录，修改 conf 下的 flume-env.sh，在里面配置 JAVA\_HOME

2、根据数据采集的需求配置采集方案，描述在配置文件中(文件名可任意自定义)

3、指定采集方案配置文件，在相应的节点上启动 flume agent 先用一个最简单的例子来测试一下程序环境是否正常

下面的内容的官网地址：[http://flume.apache.org/FlumeUserGuide.html][http_flume.apache.org_FlumeUserGuide.html]

本文使用的是最新版本的apache flume 1.8，**用最简单的单一代理Flume 配置，测试下Flume是否可以用，**

1、在$FLUME\_HOME/agentconf 目录下创建一个数据采集方案，该方案就是从一个网络端口 收集数据，也就是创一个任意命名的配置文件如下：netcat-logger.properties

**说明下，这里所有的例子都是将配置文件放到 $FLUME\_HOME/agentconf**** 目录下，后面就不赘述了。**

文件内容如下：

# 定义这个 agent 中各个组件的名字
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    # 描述和配置 source 组件：r1
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    # 描述和配置 sink 组件：k1
    a1.sinks.k1.type = logger
    # 描述和配置 channel 组件，此处使用是内存缓存的方式
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # 描述和配置 source channel sink 之间的连接关系
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

2、启动 agent 去采集数据：

在$FLUME\_HOME 下执行如下命令： **bin/flume-ng agent -c conf -f agentconf/netcat-logger.properties -n a1 - Dflume.root.logger=INFO,console**

\-c conf 指定 flume 自身的配置文件所在目录

\-f conf/netcat-logger.con 指定我们所描述的采集方案

\-n a1 指定我们这个 agent 的名字

具体参数命令请通过flume-nghelp查看

3、测试 先要往 agent 的 source 所监听的端口上发送数据，让 agent 有数据可采 例如在本机节点，使用 telnet localhost 44444 命令就可以  输入三行数据：

![70][]

4.然后会看在之前启动的终端查看console输出到如下：

![70 1][]

这里会发现消息hello world! 输出了，而hello word helloword 和 helloword hellword helloword则被拦截了。因为在配置文件中，我们选择的输出方式为：a1.sinks.k1.type= logger，即console输出，flume-ng针对**logger是只显示16个字节**的，剩下的都被sink截了。

不难看出，在event处理过程中，发生了数据截取操作。

Ok，进入下一个环节。

## 一、单一代理Flume配置 ##

![70 2][]

上面测试实例就属于单一代理flume配置，只不过source的type=netcat，一般常用source有spooldir（监视文件目录）和exec（监视文件内容），这个比较简单，可以自行测试。在这里展示一下配置文件以及说明即可

### **spoodir(监视文件目录)：** ###

**特性：** 1、监视一个目录，只要目录中出现新文件，就会采集文件中的内容

2、采集完成的文件，会被 agent 自动添加一个后缀：.COMPLETED

3、所监视的目录中不允许重复出现相同文件名的文件 下沉组件，即 sink——HDFS 文件系：hdfs sink 通道组件，即 channel——可用 file channel 也可以用内存 channel

**spooldir-hdfs.properties配置示例**：

#定义三大组件的名称
    agent1.sources = source1
    agent1.sinks = sink1
    agent1.channels = channel1
    # 配置 source 组件
    agent1.sources.source1.type = spooldir
    agent1.sources.source1.spoolDir = /home/hadoop/logs/
    agent1.sources.source1.fileHeader = false
    #配置拦截器
    agent1.sources.source1.interceptors = i1
    agent1.sources.source1.interceptors.i1.type = host
    agent1.sources.source1.interceptors.i1.hostHeader = hostname
    # 配置 sink 组件
    agent1.sinks.sink1.type = hdfs
    agent1.sinks.sink1.hdfs.path=hdfs://myha01/flume_log/%y-%m-%d/%H-%M
    agent1.sinks.sink1.hdfs.filePrefix = events
    agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
    agent1.sinks.sink1.hdfs.batchSize= 100
    agent1.sinks.sink1.hdfs.fileType = DataStream
    agent1.sinks.sink1.hdfs.writeFormat =Text
    agent1.sinks.sink1.hdfs.rollSize = 102400
    agent1.sinks.sink1.hdfs.rollCount = 1000000
    agent1.sinks.sink1.hdfs.rollInterval = 60
    #agent1.sinks.sink1.hdfs.round = true
    #agent1.sinks.sink1.hdfs.roundValue = 10
    #agent1.sinks.sink1.hdfs.roundUnit = minute
    agent1.sinks.sink1.hdfs.useLocalTimeStamp = true
    # Use a channel which buffers events in memory
    agent1.channels.channel1.type = memory
    agent1.channels.channel1.keep-alive = 120
    agent1.channels.channel1.capacity = 500000
    agent1.channels.channel1.transactionCapacity = 600
    # Bind the source and sink to the channel
    agent1.sources.source1.channels = channel1
    agent1.sinks.sink1.channel = channel1

Channel 参数解释：

capacity：默认该通道中最大的可以存储的 event 数量

trasactionCapacity：每次最大可以从 source 中拿到或者送到 sink 中的 event 数量

keep-alive：event 添加到通道中或者移出的允许时间

启动： bin/flume-ng agent -c conf -f agentconf/spooldir-hdfs.properties -n agent1

测试： 1、如果 HDFS 集群是高可用集群，那么必须要放入 core-site.xml 和 hdfs-site.xml 文件到 $FLUME\_HOME/conf 目录中 2、查看监控的/home/Hadoop/logs 文件夹中的文件是否被正确上传到 HDFS 上

3、在该目录中创建文件，或者从其他目录往该目录加入文件，验证是否新增的文件能被自 动的上传到 HDFS

### **exec（监视文件内容）：** ###

exec类型我个人记得相对简单而且最容易理解，往一个文件追加内容即可触发，如监控不断更新的日志文件，捕捉新的日志内容，后面flume的其他应用架构，都是用exec来示例。此处单一模式就列一下配置文件

**tail-hdfs.properties配置**：

agent1.sources = source1
    agent1.sinks = sink1
    agent1.channels = channel1
    # Describe/configure tail -F source1
    agent1.sources.source1.type = exec
    agent1.sources.source1.command = tail -F /home/hadoop/logs/catalina.out
    agent1.sources.source1.channels = channel1
    #configure host for source
    agent1.sources.source1.interceptors = i1
    agent1.sources.source1.interceptors.i1.type = host
    agent1.sources.source1.interceptors.i1.hostHeader = hostname
    # Describe sink1
    agent1.sinks.sink1.type = hdfs
    #a1.sinks.k1.channel = c1
    agent1.sinks.sink1.hdfs.path =hdfs://myha01/weblog/flume-event/%y-%m-%d/%H-%M
    agent1.sinks.sink1.hdfs.filePrefix = tomcat_
    agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
    agent1.sinks.sink1.hdfs.batchSize= 100
    agent1.sinks.sink1.hdfs.fileType = DataStream
    agent1.sinks.sink1.hdfs.writeFormat =Text
    agent1.sinks.sink1.hdfs.rollSize = 102400
    agent1.sinks.sink1.hdfs.rollCount = 1000000
    agent1.sinks.sink1.hdfs.rollInterval = 60
    agent1.sinks.sink1.hdfs.round = true
    agent1.sinks.sink1.hdfs.roundValue = 10
    agent1.sinks.sink1.hdfs.roundUnit = minute
    agent1.sinks.sink1.hdfs.useLocalTimeStamp = true
    # Use a channel which buffers events in memory
    agent1.channels.channel1.type = memory
    agent1.channels.channel1.keep-alive = 120
    agent1.channels.channel1.capacity = 500000
    agent1.channels.channel1.transactionCapacity = 600
    # Bind the source and sink to the channel
    agent1.sources.source1.channels = channel1
    agent1.sinks.sink1.channel = channel1

## 二、“集群”代理Flume 配置（多agent串联） ##

![SouthEast][]

严格来说flume不是分布式集群，由多个agent共同完成，但都是agent的角色，这里集群的概念是多台机器的管理，最简单的就是两台机器一台代理主机从数据源获取数据，然后将数据在传送到另一台主机上，进行输出。这样做的意义是，一个业务多数据源的时候，我们可以对每个数据源设置代理，然后将它们汇总到一台代理主机上进行输出。

下面实现最简单的集群配置，即两个代理，一台接受数据源数据的代理将数据推送到汇总的代理，而汇总的代理再将数据输出。因此这两台主机分别是push,pull

根据上图需要用AVRO RPC通信，因此推数据sinks类型与拉数据的sources的类型都是avro 。而拉数据代理的数据源，我们用前文讲的Exec Source 形式来处理，

如现在我在两台机器上的测试，192.168.2.101 和 192.168.2.102 上面做 agent 的传递：

分别是：

hadoop01：tail-avro.properties

使用 exec “tail -F /home/hadoop/testlog/date.log”获取采集数据

使用 avro sink 数据都下一个 agent

hadoop02：avro-hdfs.properties

使用 avro 接收采集数据

使用 hdfs sink 数据到目的地

**第一步准备hadoop01**

在 IP 为 192.168.2.101 的 hadoop01上的 agentconf 下创建一个 tail-avro.properties：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    # Describe/configure the source
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /home/hadoop/testlog/date.log
    a1.sources.r1.channels = c1
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.channel = c1
    a1.sinks.k1.hostname = hadoop02
    a1.sinks.k1.port = 4141
    a1.sinks.k1.batch-size = 2
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

第二步：准备hadoop02

再在 IP 为 192.168.2.102 的 hadoop02 机器上配置采集方案 avro-hdfs.properties：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    # Describe/configure the source
    a1.sources.r1.type = avro
    a1.sources.r1.channels = c1
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 4141
    # Describe k1
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path =hdfs://myha01/testlog/flume-event/%y-%m-%d/%H-%M
    a1.sinks.k1.hdfs.filePrefix = date_
    a1.sinks.k1.hdfs.maxOpenFiles = 5000
    a1.sinks.k1.hdfs.batchSize= 100
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat =Text
    a1.sinks.k1.hdfs.rollSize = 102400
    a1.sinks.k1.hdfs.rollCount = 1000000
    a1.sinks.k1.hdfs.rollInterval = 60
    a1.sinks.k1.hdfs.round = true
    a1.sinks.k1.hdfs.roundValue = 10
    a1.sinks.k1.hdfs.roundUnit = minute
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

第三步：最终测试

我们先启动pull代理。

1、首先启动pull代理，启动 hadoop02 机器上的 agent

bin/flume-ng agent -c conf -n a1 -f agentconf/avro-hdfs.properties  - Dflume.root.logger=INFO,console

2、再启动 hadoop01 上的 agent

bin/flume-ng agent -c conf -n a1 -f agentconf/tail-avro.properties  - Dflume.root.logger=INFO,console

3、执行一个普通的脚本往 hadoop01 的/home/hadoop/testlog/date.log 中追加数据，来触发监视：

#!/bin/bash 
    
    while true 
    
    do 
    
    echo `date` >> /home/hadoop/testlog/date.log 
    
    sleep 1 
    
    done

4、至此会发现在 hadoop01 agent 发送的数据会转到 hadoop02 agent，然后被 sink 到了 HDFS 的对应目录 hdfs://myha01/testlog/flume-event/

![70 3][]

## 三：多路复用采集 ##

架构设计（机器条件有限，设计两路（图中sink1和sink3）复用）：

1.第一路：hadoop01的agent直接sink到hdfs

2.第二路：hadoop01 的 flume agent 传送数据到 hadoop02的 flume agent，由hadoop02的agent来sink到hdfs：

![70 4][]

如现在我在两台机器上的测试，192.168.2.101 和 192.168.2.102 上面做 agent 的传递：

分别是： hadoop01：tail-avro.properties

使用 exec “tail -F /home/hadoop/testlog/welog.log”获取采集数据

**使用hdfs sink 数据到hdfs 并且 使用 avro sink 数据都下一个 agent （即上图中的sink1和sink3）**

hadoop02：avro-hdfs.properties 使用 avro 接收采集数据 使用 hdfs sink 数据到目的地

第一步：准备hadoop01

在 IP 为 192.168.2.101 的 hadoop01 上的 agentconf 下创建一个 tail-multiplexing.properties：

agent1.sources = source1
    agent1.sinks = sink1 sink2
    agent1.channels = channel1 channel2
    # Describe/configure tail -F source1
    agent1.sources.source1.type = exec
    agent1.sources.source1.command = tail -F /home/hadoop/logs/multiplexing.out
    agent1.sources.source1.channels = channel1 channel2
    #configure host for source
    agent1.sources.source1.interceptors = i1
    agent1.sources.source1.interceptors.i1.type = host
    agent1.sources.source1.interceptors.i1.hostHeader = hostname
    # Describe sink1
    agent1.sinks.sink1.type = hdfs
    #a1.sinks.k1.channel = c1
    agent1.sinks.sink1.hdfs.path =hdfs://myha01/weblog/flume-multiplexing/%y-%m-%d/%H-%M
    agent1.sinks.sink1.hdfs.filePrefix = tomcat_
    agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
    agent1.sinks.sink1.hdfs.batchSize= 100
    agent1.sinks.sink1.hdfs.fileType = DataStream
    agent1.sinks.sink1.hdfs.writeFormat =Text
    agent1.sinks.sink1.hdfs.rollSize = 102400
    agent1.sinks.sink1.hdfs.rollCount = 1000000
    agent1.sinks.sink1.hdfs.rollInterval = 60
    agent1.sinks.sink1.hdfs.round = true
    agent1.sinks.sink1.hdfs.roundValue = 10
    agent1.sinks.sink1.hdfs.roundUnit = minute
    agent1.sinks.sink1.hdfs.useLocalTimeStamp = true
    # Describe sink2
    agent1.sinks.sink2.type = avro
    agent1.sinks.sink2.channel = channel2
    agent1.sinks.sink2.hostname = hadoop02
    agent1.sinks.sink2.port = 4141
    agent1.sinks.sink2.batch-size = 2  
    # Use a channel which buffers events in memory
    agent1.channels.channel1.type = memory
    agent1.channels.channel1.capacity = 1000
    agent1.channels.channel1.transactionCapacity = 100
    agent1.channels.channel2.type = memory
    agent1.channels.channel2.capacity = 1000
    agent1.channels.channel2.transactionCapacity = 100
    # Bind the source and sink to the channel
    agent1.sources.source1.channels = channel1 channel2
    agent1.sinks.sink1.channel = channel1
    agent1.sinks.sink2.channel = channel2

第二步：准备hadoop02

在 IP 为 192.168.2.102 的 hadoop02 上的 agentconf 下创建一个 avro-multiplexing.properties：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    # Describe/configure the source
    a1.sources.r1.type = avro
    a1.sources.r1.channels = c1
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 4141
    # Describe k1
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path =hdfs://myha01/weblog/flume-multiplexing2/%y-%m-%d/%H-%M
    a1.sinks.k1.hdfs.filePrefix = date_
    a1.sinks.k1.hdfs.maxOpenFiles = 5000
    a1.sinks.k1.hdfs.batchSize= 100
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat =Text
    a1.sinks.k1.hdfs.rollSize = 102400
    a1.sinks.k1.hdfs.rollCount = 1000000
    a1.sinks.k1.hdfs.rollInterval = 60
    a1.sinks.k1.hdfs.round = true
    a1.sinks.k1.hdfs.roundValue = 10
    a1.sinks.k1.hdfs.roundUnit = minute
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

第三步：最终测试

1、首先启动 hadoop02 机器上的 agent

bin/flume-ng agent -c conf -n a1 -f agentconf/avro-multiplexing.properties - Dflume.root.logger=INFO,console

2、再启动hadoop01上的 agent

bin/flume-ng agent -c conf -n agent1 -f agentconf/tail-multiplexing.properties  -Dflume.root.logger=INFO,console

3、执行一个普通的脚本往 hadoop01 的/home/hadoop/logs/multiplexing.out中追加数据，触发监视：

#!/bin/bash
    while true 
    do
            echo 'multiplexing' >> /home/hadoop/logs/multiplexing.out
            sleep 1
    done

4、至此会发现

a>在 hadoop01 agent 会将数据直接sink到hdfs的对应目录hdfs://myha01/weblog/flume-multiplexing

b>同时发现hadoop01 agent 发送的数据也会转到 hadoop02 agent，

然后被 sink 到了HDFS 的对应目录 hdfs://myha01/weblog/flume-multiplexing2

![70 5][]

## 四：高可用部署采集 ##

第一步： Flume-NG 的高可用架构图：

![70 6][]

图中，我们可以看出，Flume 的存储可以支持多种，这里只列举了 HDFS 和 Kafka（如：存储 最新的一周日志，并给 Storm 系统提供实时日志流）。

第二步：节点分配 Flume 的 Agent 和 Collector 分布如下表所示：

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>名称</td> 
   <td>host</td> 
   <td>角色</td> 
  </tr> 
  <tr> 
   <td>agent1</td> 
   <td>hadoop02</td> 
   <td>日志服务器</td> 
  </tr> 
  <tr> 
   <td>agent2</td> 
   <td>hadoop03</td> 
   <td>日志服务器</td> 
  </tr> 
  <tr> 
   <td>collector1</td> 
   <td>hadoop01</td> 
   <td>AgentMaster1</td> 
  </tr> 
  <tr> 
   <td>collector2</td> 
   <td>hadoop02</td> 
   <td>AgentMaster2</td> 
  </tr> 
 </tbody> 
</table>

和架构图不同，由于条件有限，我这里只搭建了三台服务器，所以我的架构分布设计如上图，**主要实现flume的高可用即可**，图中所示，Agent1，Agent2数据分别流入到 Collector1 和 Collector2，Flume NG 本 身提供了 Failover 机制，可以自动切换和恢复。有 2个产生日志服务器分布在不 同的机房，要把所有的日志都收集到一个集群中存储。，下 面我们开发配置 Flume NG 集群

第三步：配置信息

在下面单点 Flume 中，基本配置都完成了，我们只需要新添加两个配置文件，它们是 ha\_agent.properties 和 ha\_collector.properties，其配置内容如下所示：

ha\_agent.properties 配置：

#agent name: agent1
    agent1.channels = c1
    agent1.sources = r1
    agent1.sinks = k1 k2
    #set gruop
    agent1.sinkgroups = g1
    #set channel
    agent1.channels.c1.type = memory
    agent1.channels.c1.capacity = 1000
    agent1.channels.c1.transactionCapacity = 100
    agent1.sources.r1.channels = c1
    agent1.sources.r1.type = exec
    agent1.sources.r1.command = tail -F /home/hadoop/testlog/testha.log
    agent1.sources.r1.interceptors = i1 i2
    agent1.sources.r1.interceptors.i1.type = static
    agent1.sources.r1.interceptors.i1.key = Type
    agent1.sources.r1.interceptors.i1.value = LOGIN
    agent1.sources.r1.interceptors.i2.type = timestamp
    # set sink1
    agent1.sinks.k1.channel = c1
    agent1.sinks.k1.type = avro
    agent1.sinks.k1.hostname = hadoop01
    agent1.sinks.k1.port = 52020
    # set sink2
    agent1.sinks.k2.channel = c1
    agent1.sinks.k2.type = avro
    agent1.sinks.k2.hostname = hadoop02
    agent1.sinks.k2.port = 52020
    #set sink group
    agent1.sinkgroups.g1.sinks = k1 k2
    #set failover
    agent1.sinkgroups.g1.processor.type = failover
    agent1.sinkgroups.g1.processor.priority.k1 = 10
    agent1.sinkgroups.g1.processor.priority.k2 = 1
    agent1.sinkgroups.g1.processor.maxpenalty = 10000

ha\_collector.properties 配置：

#set agent name
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    #set channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # other node,nna to nns
    a1.sources.r1.type = avro
    ## 当前主机为什么，就修改成什么主机名
    a1.sources.r1.bind = hadoop02
    a1.sources.r1.port = 52020
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = static
    a1.sources.r1.interceptors.i1.key = Collector
    ## 当前主机为什么，就修改成什么主机名
    a1.sources.r1.interceptors.i1.value = hadoop02
    a1.sources.r1.channels = c1
    #set sink to hdfs
    a1.sinks.k1.type=hdfs
    a1.sinks.k1.hdfs.path= hdfs://myha01/flume_ha/loghdfs
    a1.sinks.k1.hdfs.fileType=DataStream
    a1.sinks.k1.hdfs.writeFormat=TEXT
    a1.sinks.k1.hdfs.rollInterval=10
    a1.sinks.k1.channel=c1
    a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

注意：在把 ha\_collector.properties 文件拷贝到另外一台 collector 的时候，记得更改该配置文 件中的主机名。在该配置文件中有注释

第四步：启动 测试

先启动 hadoop01 和 hadoop02 上的 collector 角色：

bin/flume-ng agent -c conf -f agentconf/ha\_collector.properties -n a1 - Dflume.root.logger=INFO,console

然后启动 hadoop02，hadoop03 上的 agent 角色：

bin/flume-ng agent -c conf -f agentconf/ha\_agent.properties -n agent1 - Dflume.root.logger=INFO,console

在这里要提一下**flume的Failover 机制**，flume其实都是agent角色，只是指定了2个collector，正常工作collector只有一个（另一个是备份），具体哪个collector正常工作取决于ha\_agent.properties中配置的优先级

\#set failover  
agent1.sinkgroups.g1.processor.type = failover  
agent1.sinkgroups.g1.processor.priority.k1 = 10  
agent1.sinkgroups.g1.processor.priority.k2 = 1  
agent1.sinkgroups.g1.processor.maxpenalty = 10000

可以看到k1的优先级是10，比k2的优先级高，所以正常工作的是k1,在我这里也就是hadoop01,hadoop02就是备份节点，但hadoop01停止工作，collector进行切换为hadoop02工作，但flume的高可用和hadoop集群的高可用不一样的是，当hadoop01重新启动时，collector又会切换回来。从始至终都是遵循优先级原则。

所以测试高可用，先让hadoop01正常工作,hadoop02启动后只是待命，然后切断hadoop01的flume进程，此时collector是否切换，如果切换，再重新启动hadoop01，此时的collector会再次切换，这里，留一个测试。自行测试，，哈哈。。。。

**更多 Source 和 Sink 组件 （官网网址）**

更多 Sources：http://flume.apache.org/FlumeUserGuide.html\#flume-sources

更多 Channels：http://flume.apache.org/FlumeUserGuide.html\#flume-channels

更多 Sinks：http://flume.apache.org/FlumeUserGuide.html\#flume-sinks

（PS：如果博文哪里有错误，不当的知识点，望能指点！0.0）

## 五：综合案例 ##

**1.这里做一个真实的简单案例场景/需求**

A、B 两台日志服务机器实时生产日志主要类型为 access.log、nginx.log、web.log

现在要求：

把 A、B 机器中的 access.log、nginx.log、web.log 采集汇总到 C 机器上然后统一收集到 hdfs 中。

但是在 hdfs 中要求的目录为：

/source/logs/access/20160101/\*\*

/source/logs/nginx/20160101/\*\*

/source/logs/web/20160101/\*\*

**2.场景分析**

![70 7][]

**3.数据处理流程分析**

![70 8][]

**4.需求实现**

第一：准备 3 台服务器

服务器 A 对应的 IP 为 192.168.2.101，主机名为 hadoop01

服务器 B 对应的 IP 为 192.168.2.102，主机名为 hadoop02

服务器 C 对应的 IP 为 192.168.2.103，主机名为 hadoop03

第二：设计采集方案 exec\_source\_avro\_sink.properties

在服务器hadoop01和服务器hadoop02上的$FLUME\_HOME/agentconf创建采集方案的配置文件 exec\_source\_avro\_sink.properties，文件内容为：

# 指定各个核心组件
    a1.sources = r1 r2 r3
    a1.sinks = k1
    a1.channels = c1
    # 准备数据源
    ## static 拦截器的功能就是往采集到的数据的 header 中插入自己定义的 key-value 对
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /home/hadoop/logs/access.log
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = static
    a1.sources.r1.interceptors.i1.key = type
    a1.sources.r1.interceptors.i1.value = access
    a1.sources.r2.type = exec
    a1.sources.r2.command = tail -F /home/hadoop/logs/nginx.log
    a1.sources.r2.interceptors = i2
    a1.sources.r2.interceptors.i2.type = static
    a1.sources.r2.interceptors.i2.key = type
    a1.sources.r2.interceptors.i2.value = nginx
    a1.sources.r3.type = exec
    a1.sources.r3.command = tail -F /home/hadoop/logs/web.log
    a1.sources.r3.interceptors = i3
    a1.sources.r3.interceptors.i3.type = static
    a1.sources.r3.interceptors.i3.key = type
    a1.sources.r3.interceptors.i3.value = web
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop03
    a1.sinks.k1.port = 41414
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 20000
    a1.channels.c1.transactionCapacity = 10000
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sources.r2.channels = c1
    a1.sources.r3.channels = c1
    a1.sinks.k1.channel = c1

第三：准备 avro\_source\_hdfs\_sink.properties 配置文件

在服务器 C 上的$FLUME\_HOME/agentconf 中创建配置文件 avro\_source\_hdfs\_sink.properties 文件内容为：

#定义 agent 名， source、channel、sink 的名称
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    #定义 source
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port =41414
    #添加时间拦截器
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type=org.apache.flume.interceptor.TimestampInterceptor$Builder
    #定义 channels
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 20000
    a1.channels.c1.transactionCapacity = 10000
    #定义 sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path=hdfs://myha01/source/logs/%{type}/%Y%m%d
    a1.sinks.k1.hdfs.filePrefix =events
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    #时间类型
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    #生成的文件不按条数生成
    a1.sinks.k1.hdfs.rollCount = 0
    #生成的文件按时间生成
    a1.sinks.k1.hdfs.rollInterval = 30
    #生成的文件按大小生成
    a1.sinks.k1.hdfs.rollSize = 10485760
    #批量写入 hdfs 的个数
    a1.sinks.k1.hdfs.batchSize = 20
    #flume 操作 hdfs 的线程数（包括新建，写入等）
    a1.sinks.k1.hdfs.threadsPoolSize=10
    #操作 hdfs 超时时间
    a1.sinks.k1.hdfs.callTimeout=30000
    #组装 source、channel、sink
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

第四：启动

配置完成之后，在服务器 A 和 B 上的/home/hadoop/logs 有数据文件 access.log、nginx.log、 web.log。

4.1先启动服务器 C（hadoop03）上的 flume，在 flume 安装目录下执行启动命令：

bin/flume-ng agent -c conf -f agentconf/avro\_source\_hdfs\_sink.properties -name a1 - Dflume.root.logger=DEBUG,console

4.2然后在启动服务器上的 A（hadoop01）和 B（hadoop02），在 flume 安装目录下执 行启动命令：

bin/flume-ng agent -c conf -f agentconf/exec\_source\_avro\_sink.properties -name a1 - Dflume.root.logger=DEBUG,console

4.3编写一个脚本文件test\_dome.sh向日志文件追加数据：

#!/bin/bash
    while true
    do
            echo 'access' >> /home/hadoop/logs/access.log
            echo 'nginx' >> /home/hadoop/logs/nginx.log
            echo 'web'  >> /home/hadoop/logs/web.log
            sleep 2
    done

第五：测试

执行test\_dome.sh脚本，查看hdfs目录

![70 9][]

这里就示例一个，进入access文件夹

![70 10][]

查看输出日志文件

hadoop fs -cat /source/logs/access/20180813/events.1534147475666

![70 11][]

hadoop fs -cat /source/logs/nginx/20180813/events.1534147410509

![70 12][]

结果如预期一般，实现了需求，至此这个简单的实战案例就完成了。

[http_flume.apache.org_FlumeUserGuide.html]: http://flume.apache.org/FlumeUserGuide.html
[70]: /images/20220516/20d0b8712ddd474f8de2468bff47ad06.png
[70 1]: /images/20220516/d99814b995ae4fe29f1f3bbbb1b34431.png
[70 2]: /images/20220516/46527c78e4154a5f8c76942032c423d6.png
[SouthEast]: /images/20220516/8c926477cd9e4461bc585850e6fff379.png
[70 3]: /images/20220516/729d06a6f7234ce9a817ae27eccb8a65.png
[70 4]: /images/20220516/cedfed70b7024584b5a97a8d63c1a1ba.png
[70 5]: /images/20220516/5945bbe1ab124111a28271820dd00751.png
[70 6]: /images/20220516/1f419139edf844e7aa75da2f1a54efea.png
[70 7]: /images/20220516/bdccabb111c04d3aaa0618228a012d80.png
[70 8]: /images/20220516/7c57fbe2b2b84831a802b8c1f36c9416.png
[70 9]: /images/20220516/1f434c57ee3f47d5a43b2f4b38641c9f.png
[70 10]: /images/20220516/0cd3f752d1ec4011a6a963ac77a07aea.png
[70 11]: /images/20220516/a621c34edf694a0299077f4a568b4d70.png
[70 12]: /images/20220516/3c84bbb2b0f54ab5ab5e058ad45c5881.png