Hadoop高手之路8-Flume日志采集

Hadoop高手之路8-Flume日志采集

在大数据系统的开发中，数据收集工作无疑是开发者首要解决的一个难题，但由于生产数据的源头丰富多样，其中包含网站日志数据、后台监控数据、用户浏览网页数据等，数据工程师要想将它们分门别类的采集到HDFS系统中，就可以使用Apache Flume（数据采集）系统。

一、Flume概述

1. Flume简介

Flume原是Cloudera公司提供的一个高可用的、高可靠的、分布式海量日志采集、聚合和传输系统，而后纳入到了Apache旗下，作为一个顶级开源项目。Apache Flume不仅只限于日志数据的采集，由于Flume采集的数据源是可定制的，因此Flume还可用于传输大量事件数据，包括但不限于网络流量数据、社交媒体生成的数据、电子邮件消息以及几乎任何可能的数据源。

2. Flume运行机制

Flume的核心是把数据从数据源（例如Web服务器）通过数据采集器（Source）收集过来，再将收集的数据通过缓冲通道（Channel）汇集到指定的接收器（Sink）。

Flume基本架构中有一个Agent（代理），它是Flume的核心角色，Flume Agent是一个JVM进程，它承载着数据从外部源流向下一个目标的三个核心组件：Source、Channel和Sink。

3. Flume日志采集系统结构图

在实际开发中， Flume需要采集数据的类型多种多样，同时还会进行不同的中间操作，所以根据具体需求，可以将Flume日志采集系统分为简单结构和复杂结构。

简单的Flume日志采集系统的结构

复杂的Flume日志采集系统的结构

二、Flume的搭建

1. 下载

注意：使用1.9.0这个版本

2. 上传

3. 解压

4. 配置环境变量

5. 配置flume

修改flume-env.sh文件

三、Flume入门使用

1. 配置数据采集方案

1) 查看官网

2) 案例需求

flume连接和监听服务器的某个端口，采集数据并显示

3) 创建新的配置文件

4) 复制官网的采集配置示例，在此基础上进行修改

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = logger

# 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

2. 启动flume进行采集

启动命令查看官网

flume-ng agent --conf conf --conf-file conf/netcat-logger.conf --name a1 -Dflume.root.logger=INFO,console

3. 采集数据测试

用telnet向本机的44444端口发送数据，模拟服务器产生数据

在使用之前，要先安装telnet

在telnet端发送数据

在flume端显示采集到的数据

四、Flume采集方案说明

1. Flume Sources

在编写Flume采集方案时，首先必须明确采集的数据源类型、出处；接着，根据这些信息与Flume已提供支持的Flume Sources进行匹配，选择对应的数据采集器类型（即sources.type）；再根据选择的数据采集器类型，配置必要和非必要的数据采集器属性，Flume提供并支持的Flume Sources种类如下所示。

1) Avro Source

监听Avro端口并从外部Avro客户端流中接收event数据，当与另一个Flume Agent上的Avro Sink配对时，可创建分层集合拓扑，利用Avro Source可以实现多级流动、扇出流、扇入流等效果，Avro Source常用配置属性如下。

2) Spooling Directory Source

Spooling Directory Source允许对指定磁盘上的文件目录进行监控来提取数据，它将查看文件的指定目录的新增文件，并将文件中的数据读取出来。Spooling Directory Source常用配置属性如下表所示。

3) Taildir Source

Taildir Source用于观察指定的文件，几乎可以实时监测到添加到每个文件的新行。如果文件正在写入新行，则此采集器将重试采集它们以等待写入完成，Source常用配置属性如下所示。

4) HTTP Source

HTTP Source可以通过HTTP POST和GET请求方式接收event数据，GET通常只能用于测试使用，POST请求发送的所有的events都被认为是一个批次，会在一个事务中插入channel，HTTP Source常用配置属性如下所示。

2. Flume Channels

​ Channels通道是event在Agent上暂存的存储库，Source向Channel中添加event，Sink在读取完数据后再删除它。在配置Channels时，需要明确的就是将要传输的sources数据源类型；根据这些信息结合开发中的实际需求，选择Flume已提供的支持的Flume Channels；再根据选择的Channel类型，配置必要和非必要的Channel属性，Flume提供并支持的Flume Channels种类如下所示。

1) Memory Channel

Memory Channel会将event存储在具有可配置最大尺寸的内存队列中，适用于需要更高吞吐量的流量，但在Agent发生故障时会丢失部分阶段数据，下表为Memory Channel常用配置属性。

2) File Channel

File Channel是Flume的持久通道，它将所有event写入磁盘，因此不会丢失进程或机器关机、崩溃时的数据。File Channel通过在一次事务中提交多个event来提高吞吐量，做到了只要事务被提交，那么数据就不会有丢失，File Channel常用配置属性如下所示。

3. Flume Sinks

Flume Soures采集的数据通过Channels通道流向Sink中，此时Sink类似一个集结的递进中心，它需要根据需求进行配置，从而最终选择发送目的地。配置Sinks时，明确将要传输的数据目的地、结果类型；然后根据实际需求信息，选择Flume已提供支持的Flume Sinks；再根据选择的Sinks类型，配置必要和非必要的Sinks属性。Flume提供并支持的Flume Sinks种类如下所示。

1) HDFS Sink

HDFS Sink将event写入Hadoop分布式文件系统（HDFS），它目前支持创建文本和序列文件，以及两种类型的压缩文件，下表为HDFS Sink常用配置属性。

2) Logger Sink

Logger Sink用于记录INFO级别event，它通常用于调试。Logger Sink接收器的不同之处是它不需要在“记录原始数据”部分中说明额外的配置，Logger Sink常用配置属性如下所示。

3) Avro Sink

Avro Sink形成Flume分层收集支持的一半，发送到此接收器的Flume event转换为Avro event，并发送到对应配置的主机名/端口，event将从配置的channel中批量获取配置的批处理大小，Avro Sink常用配置属性如下所示。

五、Flume采集数据案例一——监控文件夹的变化

1. 需求

监控服务器下的某个文件夹（日志输出文件夹），如果该文件夹下有新的文件，则采集该文件的内容上传到hadoop集群上。

2. 编写数据采集配置方案

source的官网示例

sink的官网示例

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字及各个组件sources、sinks、channels
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 配置数据源
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /var/log/flumeSpool
a1.sources.r1.fileHeader = true

# Describe the sink
# 描述数据的目的地（下沉）
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%Y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute

# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/spoolingdir-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

3. 启动hdfs集群

4. 启动flume

出现错误，提示监控目录不存在，需要提前创建

再次启动flume

5. 测试flume数据采集

在/var/log/flumeSpool文件夹下添加一个新文件

出现错误，修改采集方案

再次启动flume测试

提示错误

用hadoop的guava新版本替换flume的旧版本

再次启动flume，然后在/var/log/flumeSpool文件夹下添加一个新文件

在hdfs集群上查看

解决乱码，修改文件：

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字及各个组件sources、sinks、channels
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 配置数据源
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /var/log/flumeSpool
a1.sources.r1.fileHeader = true

# Describe the sink
# 描述数据的目的地（下沉）
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%Y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true

a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.fileType = DataStream

# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/spoolingdir-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

重新启动flume，在创建一个新的文件

这样结果就可以了。

六、Flume采集数据案例二——监控文件的变化

1.采集方案

数据源

数据下沉不要修改，采集配置方案如下：

配置代码：

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 定义数据源
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/text.log


# Describe the sink
# 定义数据的目的地（下沉）
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
# 是否循环创建文件夹
a1.sinks.k1.hdfs.round = true
# 循环创建文件夹的时间间隔是10分钟
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
# 使用本地时间格式
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 时间间隔
a1.sinks.k1.hdfs.rollInterval = 3
# 大小间隔
a1.sinks.k1.hdfs.rollSize = 20
# event的个数,这三个参数谁先满足就出发循环滚动
a1.sinks.k1.hdfs.rollCount = 5
# 批处理数量
a1.sinks.k1.hdfs.batchSize = 1
# 文件格式,表示普通文本格式
a1.sinks.k1.hdfs.fileType = DataStream 

# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/exec-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

2. 测试采集功能

模拟场景
先写一个shell脚本，持续输出当前日期到监控文件/var/log/test.log中，模拟服务器日志的文件

while true;do date >> /var/log/text.log;done

3. 再克隆一个会话，查看新增的内容

4. 启动flume

flume-ng agent --conf conf --conf-file conf/exec-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

5. 查看HDFS上的结果

七、Flume的可靠性保证-负载均衡

1. 概念

配置的采集方案是通过唯一一个Sink作为接收器接收后续需要的数据，但会出现当前Sink故障或数据收集请求量较大的情况，这时单一Sink配置可能就无法保证Flume开发的可靠性。因此，Flume 提供Flume Sink Processors解决上述问题。

Sink处理器允许定义Sink groups，将多个sink分组到一个实体中，Sink处理器就可通过组内多个sink为服务提供负载均衡功能。

负载均衡接收器处理器（Load balancing sink processor）提供了在多个sink上进行负载均衡流量的功能，它维护一个活跃的sink索引列表，需在其上分配负载，还支持round_robin（轮询）和random（随机）选择机制进行流量分配，默认选择机制为round_robin。Load balancing sink processor提供的配置属性如下所示。

2. 搭建并配置flume集群

三台服务器的flume集群：hadoop01、hadoop02、hadoop03

1) 分发hadoop01上的flume文件到hadoop02和hadoop03上

2) 分发环境变量配置文件

3) 使环境变量起作用

3. 配置采集方案

两级配置方案

查看官方文档的示例

1) 在hadoop001上配置第一级采集方案

在conf下编写采集方案exec-avro.conf

代码如下：

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2

#定义组的属性
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = load_balance

#定义负载均衡
a1.sinkgroups.g1.processor.backoff = true
a1.sinkgroups.g1.processor.selector = random
a1.sinkgroups.g1.processor.selector.maxTimeOut = 30000


# Describe/configure the source
# 定义数据源
a1.sources.r1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/test1.log

# Describe the sink
# 定义数据的目的地1（下沉）
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = hadoop002
a1.sinks.k1.port = 4545

# 定义数据的目的地2（下沉）
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c1
a1.sinks.k2.hostname = hadoop003
a1.sinks.k2.port = 4545


# Use a channel which buffers events in memory
# 定义管道
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 启动命令
# flume-ng agent --conf conf --conf-file conf/exec-avro.conf --name a1 -Dflume.root.logger=INFO,console

2）在hadoop002和hadoop003上配置第二级采集方案

hadoop002

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 定义数据源
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop002
a1.sources.r1.port = 4545
# Describe the sink

# 定义数据的目的地（下沉）
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
# 是否循环创建文件夹
a1.sinks.k1.hdfs.round = true
# 循环创建文件夹的时间间隔是十分钟
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
# 使用本地时间个数
a1.sinks.k1.hdfs.useLocalTimeStamp = true
// 列编辑模式，按住alt选择多列

# 时间间隔
a1.sinks.k1.hdfs.rollInterval = 3
# 大小间隔
a1.sinks.k1.hdfs.rollSize = 20
# event的个数，这三个参数谁先满足就出发循环滚动
a1.sinks.k1.hdfs.rollCount = 5
# 批处理数量
a1.sinks.k1.hdfs.batchSize = 1 
# 文件格式 表示普通文本文件
a1.sinks.k1.hdfs.fileType = DataStream


# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/avro-hdfs.conf --name a1 -Dflume.root.logger=INFO,console
# flume-ng agent --conf conf --conf-file avro-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

hadoop003

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 定义数据源
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop003
a1.sources.r1.port = 4545
# Describe the sink

# 定义数据的目的地（下沉）
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
# 是否循环创建文件夹
a1.sinks.k1.hdfs.round = true
# 循环创建文件夹的时间间隔是十分钟
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
# 使用本地时间个数
a1.sinks.k1.hdfs.useLocalTimeStamp = true
// 列编辑模式，按住alt选择多列

# 时间间隔
a1.sinks.k1.hdfs.rollInterval = 3
# 大小间隔
a1.sinks.k1.hdfs.rollSize = 20
# event的个数，这三个参数谁先满足就出发循环滚动
a1.sinks.k1.hdfs.rollCount = 5
# 批处理数量
a1.sinks.k1.hdfs.batchSize = 1 
# 文件格式 表示普通文本文件
a1.sinks.k1.hdfs.fileType = DataStream


# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/avro-hdfs.conf --name a1 -Dflume.root.logger=INFO,console
# flume-ng agent --conf conf --conf-file avro-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

4. 启动flume

1) 在hadoop002和hadoop003上flume

从最后一级开始启动flume
hadoop002

hadoop003

2) 在hadoop001上启动flume

同时在hadoop002和hadoop003上也提示连接成功

5. 负载均衡的测试

1) 克隆hadoop001的会话，编写脚本并运行

while true;do date >> /var/log/test1.log;sleep 5;done

2) 克隆hadoop001的会话，查看文件变化

tail -F /var/log/test1.log

6. 查看结果

1）hadoop002

2）hadoop003

3) 在hadoop集群上

八、Flume的可靠性保证-故障恢复

1. 概念

故障转移接收器处理器（Failover Sink Processor）维护一个具有优先级的sink列表，保证在处理event时，只需有一个可用的sink即可。

故障转移机制工作原理是将故障的sink降级到故障池中，在池中为它们分配一个冷却期，在重试之前冷却时间会增加，当sink成功发送event后，它将恢复到活跃池中。Failover Sink Processor提供的配置属性如下所示。

2. 配置采集方案

只需要改动第一级采集方案部分内容

改为故障恢复:

代码如下：

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2

#定义组的属性
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000


# Describe/configure the source
# 定义数据源
a1.sources.r1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/test1.log

# Describe the sink
# 定义数据的目的地1（下沉）
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = hadoop002
a1.sinks.k1.port = 4545

# 定义数据的目的地2（下沉）
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c1
a1.sinks.k2.hostname = hadoop003
a1.sinks.k2.port = 4545


# Use a channel which buffers events in memory
# 定义管道
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 启动命令
# flume-ng agent --conf conf --conf-file conf/exec-avro-failover.conf --name a1 -Dflume.root.logger=INFO,console

注意：启动命令也要做对应的修改。

第二级采集方案不用修改。

3. 启动flume

从最后一级开始启动flume。

启动hadoop002和hadoop003：

由于hadoop003的优先级比hadoop002高，所以hadoop003先工作，hadoop002处于待机状态。

启动hadoop001

4. 启动测试脚本

克隆hadoop001的会话，运行测试脚本。

while true;do date >> /var/log/test1.log;sleep 5;done

5. 故障恢复

把hadoop03的flume关闭，等待十秒钟（超时时间，在采集方案中进行的定义）

查看hadoop集群

九、Flume拦截器

1. 概念

Flume Interceptors（拦截器）用于对Flume系统数据流中event的修改操作。使用Flume拦截器时，只需参考官方配置属性在采集方案中选择性的配置即可，当涉及到配置多个拦截器时，拦截器名称间需用空格分隔，且拦截器配置顺序就是拦截顺序。Flume 1.8.0版本中，Flume提供并支持的拦截器有很多，具体如下所示。

1) Timestamp Interceptor

Timestamp Interceptor（时间戳拦截器）将流程执行时间插入到event的header头部，此拦截器插入带有timestamp键的标头，其值为对应时间戳。若配置中已存在时间戳时，此拦截器可保留现有时间戳，Timestamp Interceptor提供的常用配置属性如下所示。

2) Static Interceptor

Static Interceptor（静态拦截器）允许用户将具有静态值的静态头附加到所有event。当前不支持一次指定多个header头，但是用户可定义多个Static Interceptor来为每一个拦截器都追加一个header，Static Interceptor提供的常用配置属性如下所示。

2. 场景

在实际开发的应用场景中，两台服务器A、B在实时产生日志数据，日志类型主要为access.log、nginx.log和web.log。现需要将A、B两台服务器产生的日志数据access.log、nginx.log和web.log采集汇总到服务器C上，并统一收集并上传到HDFS文件系统进行保存。在HDFS中保存日志数据的文件必须按照以下要求进行归类统计（20180723表示收集日志数据的当前日期）：

• /source/logs/access/20180723/**
• /source/logs/nginx/20180723/**
• /source/logs/web/20180723/**

3. 日志数据采集流程图

根据案例需求启动3台服务器，同时搭建Flume系统和Hadoop集群。此案例将hadoop002和hadoop003分别作为A服务器和B服务器进行第一阶段的日志数据采集，将hadoop001作为C服务器进行日志数据汇总并上传至HDFS。

4. hadoop002和hadoop003的配置文件

在hadoop002和hadoop003的conf目录下编写相同日志采集方案exec-avro_logCollection.conf。

# example.conf: A single-node Flume configuration

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1 r2 r3
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 定义r1数据源

a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/access.log

# 定义拦截器r1
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

# 定义r2数据源

a1.sources.r2.type = exec
a1.sources.r2.command = tail -F /var/log/nginx.log

# 定义拦截器r2
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

# 定义r3数据源

a1.sources.r3.type = exec
a1.sources.r3.command = tail -F /var/log/web.log

# 定义拦截器r3
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 = hadoop001
a1.sinks.k1.port = 41414



# 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.sources.r2.channels = c1
a1.sources.r3.channels = c1
a1.sinks.k1.channel = c1

# 启动命令
# flume-ng agent --conf conf --conf-file conf/exec-avro_logCollection.conf --name a1 -Dflume.root.logger=INFO,console

在hadoop002上创建

在hadoop003上创建

5. hadoop001的配置文件

在hadoop001机器的conf目录下编写第二级日志采集方案avro-hdfs_logCollection.conf

# Name the components on this agent
# 定义代理的名字a1及各个组件sources、sinks和channels

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 定义数据源
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop001
a1.sources.r1.port = 41414

#定义拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp

# Describe the sink
# 定义数据的目的地（下沉）

a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/logs/%{type}/%y%m%d
a1.sinks.k1.hdfs.filePrefix = events-
# 是否循环创建文件夹
a1.sinks.k1.hdfs.round = true
# 循环创建文件夹的时间间隔是十分钟
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
# 使用本地时间个数
a1.sinks.k1.hdfs.useLocalTimeStamp = true
// 列编辑模式，按住alt选择多列

# 时间间隔
a1.sinks.k1.hdfs.rollInterval = 0
# 大小间隔
a1.sinks.k1.hdfs.rollSize = 10485760
# event的个数，这三个参数谁先满足就出发循环滚动
a1.sinks.k1.hdfs.rollCount = 0
# 批处理数量
a1.sinks.k1.hdfs.batchSize = 10 
# 文件格式 表示普通文本文件
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.threadsPoolSize=10
a1.sinks.k1.hdfs.callTimeout=30000


# 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

# 启动命令
# flume-ng agent --conf conf --conf-file conf/avro-hdfs_logCollection.conf --name a1 -Dflume.root.logger=INFO,console

6. 启动flume

1) 先启动hadoop01上的flume

2) 启动hadoop002和hadoop003

启动hadoop002

启动hadoop003

3) 查看hadoop001

7. 测试效果

1) 执行脚本文件

在hadoop002和hadoop003上分别克隆三个会话，分别执行以下三个脚本，用来产生生产日志数据：

while true;do echo "access access ……"  >> /var/log/access.log;sleep 5;done

while true;do echo "nginx nginx ……"  >> /var/log/nginx.log;sleep 5;done

while true;do echo "web web ……"  >> /var/log/web.log;sleep 5;done

在窗口中分别执行上述指令后，会不断循环产生数据，为了后续更好查看效果，执行一会后就可以直接关停上述三个指令。

2) 查看hadoop001结果

关闭上述指令后查看hadoop001

3) 查看hadoop集群

文章来源地址https://www.toymoban.com/news/detail-755823.html

到了这里，关于Hadoop高手之路8-Flume日志采集的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！