Flume

Flume 简介及核心概念

Posted by leone on 2018-10-18

Flume

Flume 简介及核心概念

什么是 Flume

Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,目前是Apache的顶级项目。Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能力。

Flume 优点

1、可靠性
当节点出现故障时,日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障,从强到弱依次分别为:

  • end-to-end(收到数据agent首先将event写到磁盘上,当数据传送成功后,再删除;如果数据发送失败,可以重新发送。),
  • Store on failure(这也是scribe采用的策略,当数据接收方crash时,将数据写到本地,待恢复后,继续发送),
  • Best effort(数据发送到接收方后,不会进行确认)。

2、可扩展性
Flume采用了三层架构,分别为agent,collector和storage,每一层均可以水平扩展。
其中,所有agent和collector由master统一管理,这使得系统容易监控和维护,且master允许有多个(使用ZooKeeper进行管理和负载均衡),这就避免了单点故障问题。

3、可管理性

  • 所有agent和colletor由master统一管理,这使得系统便于维护。
  • 多master情况,Flume利用ZooKeeper和gossip,保证动态配置数据的一致性。
  • 用户可以在master上查看各个数据源或者数据流执行情况,且可以对各个数据源配置和动态加载。
  • Flume提供了web 和shell script command两种形式对数据流进行管理。

4、功能可扩展性

  • 用户可以根据需要添加自己的agent,collector或者storage。
  • 此外,Flume自带了很多组件,包括各种agent(file, syslog等),collector和storage(file,HDFS等)。

5、文档丰富,社区活跃
Flume 已经成为 Hadoop 生态系统的标配,它的文档比较丰富,社区比较活跃,方便我们学习。

flume agent

Flume agent每个agent是一个独立的Java进程,从客户端(其他agent)接收数据然后转发到下一个destination(sink(沉槽) | agent)
Agent包含三个组件:

  • A. Source(源)->生成数据的地方

从事件生成器接收数据,以event事件的形式传给一个或多个channel

  • B. Channel(通道)

从source中接受flume event,作为临时存放地,缓存到buffer中,直到sink
将其消费掉,是source和sink之间的桥梁
Channel是事务的,可以和多个source或sink协同

  • C.sink(沉槽)

存放数据到HDFS,从channel中消费event,并分发给destination,sink的
Destination 也可以是另一个agent或者HDFS,HBASE
注意:一个flume的agent,可以有多个source,channel,sink

flume核心组件介绍

Source: 完成对日志数据的收集,分成transtion 和 event 打入到channel之中, Flume提供了各种source的实现,包括Avro Source、 Exce Source、 Spooling
Directory Source、 NetCat Source、 Syslog Source、 Syslog TCP Source、Syslog UDP Source、 HTTP Source、 HDFS Source, etc。

Channel: Channel用于连接Source和Sink,Source将日志信息发送到Channel,Sink从Channel消费日志信息;Channel是中转日志信息的一个临时存储,保存有Source组件传递过来的日志信息。, flume提供了Memory Channel、 JDBC Chanel、 File Channel,etc

Sink: Flume Sink取出Channel中的数据,进行相应的存储文件系统,数据库,或者提交到远程服务器。包括HDFS sink、 Logger sink、 Avro sink、 File Roll sink、 Null sink、 HBasesink, etc。

flume安装

flume 几乎不用添加额外的配置只自定义配置好相关的配置启动即可

1
2
3
4
5
6
7
8
# 下载
$ wget http://mirrors.tuna.tsinghua.edu.cn/apache/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz

# 解压
$ tar –zxvf apache-flume-1.7.0-bin.tar.gz

# 启动flume日志打印到控制台
$FLUME_HOME/bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1 -Dflume.root.logger=INFO,console
  • exec-hdfs.conf (监控指定文件写入到hdfs中)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# 定义 agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 定义 source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/logs/web.log

# 定义 sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://node-1:9000/hadoop-2.7.7/input1/%y-%m-%d/%H-%M
a1.sinks.k1.hdfs.filePrefix = app-log
a1.sinks.k1.hdfs.fileSuffix = .dat
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 = 10240
a1.sinks.k1.hdfs.rollInterval = 120

a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 1
a1.sinks.k1.hdfs.roundUnit = minute

a1.sinks.k1.hdfs.useLocalTimeStamp = true

# 定义channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 绑定source 和 sink 到channels
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
  • exec-kafka.conf (监控指定文件写入到kafka中)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# 定义 agent
agent.sources = s1
agent.channels = c1
agent.sinks = k1

# 定义 source
agent.sources.s1.type = exec
agent.sources.s1.command = tail -F /root/logs/web.log
agent.sources.s1.channels = c1
agent.channels.c1.type = memory

# 定义 channel
agent.channels.c1.capacity = 10000
agent.channels.c1.transactionCapacity = 100


# 定义 sink
# 设置Kafka接收器
agent.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# 设置Kafka的broker地址和端口号
agent.sinks.k1.brokerList = node-2:9092, node-3: 9092, node-3: 9094
# 设置Kafka的Topic
agent.sinks.k1.topic = log-topic
# 设置序列化方式
agent.sinks.k1.serializer.class = kafka.serializer.StringEncoder

# 绑定source 和 sink 到channels
agent.sinks.k1.channel = c1
agent.source.s1.channle = c1
  • spooldir-hdfs.conf (监控指定目录写入到hdfs中)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 定义 agent
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 定义 source
a1.sources.r1.type = spooldir
# 设置要扫描的文件夹
a1.sources.r1.spoolDir = /root/logs/web
# 设置以.log结尾的文件不扫描
a1.sources.r1.ignorePattern = ^(.)*\\.log$
# 设置扫描完成的文件加一个后缀
a1.sources.r1.fileSuffix = .delete

# 定义 channels 类型可以是Memory, JDBC, File, Psuedo Transaction 比较常见的是前三种
a1.channels.c1.type = memory
# a1.channels.c1.checkpointDir = /root/filechannel/checkpoint
# a1.channels.c1.dataDirs = /root/filechannel/data

# 定义 sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://node-1:9000/flume-1.7.0/logs/%Y%m%d
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.batchSize = 100
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.filePrefix = app-log
a1.sinks.k1.hdfs.fileSuffix = .dat

# 每隔多长时间滚动一次(秒)
a1.sinks.k1.hdfs.rollInterval = 120
# 文件多大滚动一次(bytes) 512kb
a1.sinks.k1.hdfs.rollSize = 0
# 写入多少个event数据后滚动文件(写多少行就滚动)
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.minBlockReplicas = 1


# 绑定 source 和 channel 到 agent
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
  • spooldir-kafka.conf (监控指定目录写入到kafka中)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# 定义 agent
a1.sources = r3
a1.channels = c3
a1.sinks = k3

# 定义 source
a1.sources.r3.type = spooldir
# 设置要扫描的文件夹
a1.sources.r3.spoolDir = /root/logs/
# 设置以.log结尾的文件不扫描
a1.sources.r3.ignorePattern = ^(.)*\\.log$
# 设置扫描完成的文件加一个后缀
a1.sources.r3.fileSuffix = .delete

# 定义个 channels 类型可以是Memory, JDBC, File, Psuedo Transaction 比较常见的是前三种
a1.channels.c3.type = file
a1.channels.c3.checkpointDir = /root/filechannel/checkpoint
a1.channels.c3.dataDirs = /root/filechannel/data

# 定义 sink
a1.sinks.k3.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k3.brokerList = node-2:9092, node-3: 9092, node-3: 9094
a1.sinks.k3.topic = log-topic
a1.sinks.k3serializer.class = kafka.serializer.StringEncoder

# 绑定 source 和 channel
a1.sources.r3.channels = c3
a1.sinks.k3.channel = c3