Flume 核心概念及快速入门

ゞ浴缸里的玫瑰 2022-03-31 07:42 430阅读 0赞

# 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 几乎不用添加额外的配置只自定义配置好相关的配置启动即可

# 下载
    $ 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中)

# 定义 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中)

# 定义 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中)

# 定义 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中)

# 定义 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