【数据采集工具】Flume从入门到面试学习总结

国科大学习生活（期末复习资料、课程大作业解析、大厂实习经验心得等）: 文章专栏（点击跳转）
大数据开发学习文档（分布式文件系统的实现，大数据生态圈学习文档等）: 文章专栏（点击跳转）

【数据采集工具】Flume从入门到面试学习总结

- 1. Flume概述
- - 1.1 什么是Flume？
  - 1.2 Flume基础架构
- 2. Flume进阶内容
- - 2.1 Flume事务
  - 2.2 Flume Agent 内部原理
  - 2.3 Flume 拓扑结构
  - 2.4 Flume 企业开发案例
- 3. Flume生产经验（重点）
- - 3.1 Flume 参数调优
  - 3.2 Flume 采集数据会丢失吗?
  - 3.3 Flume与Kafka
- 参考文献
- 项目地址

1. Flume概述

1.1 什么是Flume？

Flume 是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。Flume基于流式架构，灵活简单。

在这里插入图片描述

分布式：Flume可以运行在多个节点上，支持数据在节点间流动。
可扩展性：Flume可以轻松扩展以处理从几百KB到几PB的数据。
可靠性：Flume提供了数据不丢失的机制，即使在节点故障的情况下也能确保数据的完整性。
有效性：Flume提供了多种数据源和数据接收器，可以有效地从各种数据源收集数据，并将其传输到不同的数据接收器。
灵活性：Flume允许用户自定义数据源和数据接收器，以适应不同的数据收集和传输需求。

1.2 Flume基础架构

在这里插入图片描述

Agent：一个JVM进程，它以事件的形式将数据从源头送至目的。Agent 主要有3个部分组成，Source、Channel、Sink。

Source：数据源，负责从外部系统收集数据，Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、taildir、sequence generator、syslog、http、legacy。
Channel：位于Source 和Sink之间的缓冲区。因此，Channel允许Source和Sink运作在不同的速率上。Channel 是线程安全的，可以同时处理几个Source 的写入操作和几个 Sink 的读取操作。Flume 自带两种Channel：Memory Channel 和 File Channel。
- Memory Channel ：内存中的队列。Memory Channel在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么Memory Channel就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。
- File Channel：将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。
Sink：数据接收器，会不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。

Event：数据传输单元，Flume以 Event 的形式将数据从源头送至目的地。 Event 由Header 和Body 两部分组成，Header用来存放该event的一些属性，为K-V结构， Body 用来存放该条数据，形式为字节数组。
在这里插入图片描述

2. Flume进阶内容

2.1 Flume事务

在 Apache Flume 中，事务是一种确保数据在不同组件之间可靠传递的机制。

在这里插入图片描述

由上图可以看出Flume Agent中Sources和Sinks分别在事务中封装了事件的存储/检索，这些事务是由Channel提供的。这确保了事件集合能够可靠地从一个点传递到另一个点。

Source端Put事务流程
doPut：将批数据先写入临时缓冲区putList
doCommit：检查Channel内存队列是否足够合并
doRollback：Channel内存队列空间不足，回滚数据（此处可能会丢数据）
Sink端Take事务流程
doTake：将数据取到临时缓冲区takeList，并将数据发送到外部文件系统（如HDFS）或另一个Flume Agent
doCommit：如果数据全部发送成功，则清除临时缓冲区takeList
doRollback：数据发送过程中如果出现异常，rollback将临时缓冲区takeList（双端队列）中的数据归还给Channel内存队列（双端队列）（此处可能会导致重复数据产生）

2.2 Flume Agent 内部原理

数据在Flume流动示意图：

在这里插入图片描述

总体流程:

外部数据被Source监听获取后，会发往ChannelProcessor处理事件（Event）；
此时数据并不会被直接发往Channel，而是先由Interceptors拦截器链对数据做一些预处理，然后再发往ChannelSelector选出事件（Event）将要被发往哪个Channel；
ChannelSelector会返回事件Channel列表给ChannelProcessor，然后才根据事件Channel列表将事件发往相应的Channel
数据到达Channel后并不是直接发往下游Sink的，而是由SinkProcessor处理后决定发往哪个Sink。

重要组件：

ChannelSelector
ChannelSelector 的作用就是选出Event将要被发往哪个Channel。其共有两种类型，分别是Replicating（复制）和Multiplexing（多路复用）。
- Replicating Selector 会将同一个Event发往所有的Channel（类似与广播）
- Multiplexing 会根据相应的原则，将不同的Event发往不同的Channel
SinkProcessor
SinkProcessor 共有三种类型，分别是 DefaultSinkProcessor （对应的是单个的 Sink）、
LoadBalancingSinkProcessor 和 FailoverSinkProcessor 对应的是Sink Group，LoadBalancingSinkProcessor 可以实现负载均衡的功能，FailoverSinkProcessor可以实现错误恢复的功能（通过配置Sink优先级实现）。

2.3 Flume 拓扑结构

2.3.1 简单串联
在这里插入图片描述

这种模式是将多个flume顺序连接起来了，从最初的source开始到最终sink 传送到目的存储系统。此模式不建议桥接过多的flume数量，因为flume数量过多不仅会影响传输速率，而且一旦传输过程中某个节点flume宕机，会影响整个传输系统。

2.3.2 复制和多路复用
在这里插入图片描述

Flume 支持将事件流向一个或者多个目的地。这种模式可以将相同数据**复制（广播）**到多个 channel 中（Channel Selector使用Replicating），或者将不同数据分发到不同的 channel 中（Channel Selector使用Multiplexing 并配合自定义Interceptor），Sink 可以选择传送到不同的目的地。
具体实现见下文Flume 企业开发案例

2.3.3 负载均衡和故障转移
在这里插入图片描述

Flume支持使用将多个Sink逻辑上分到一个Sink组，Sink组配合不同的SinkProcessor 可以实现负载均衡（SinkProcessor使用LoadBalancingSinkProcessor）和错误恢复（SinkProcessor使用FailoverSinkProcessor并在配置文件中设置各个Sink的优先级）的功能。

2.3.4 聚合
在这里插入图片描述

这种模式是我们最常见的，也非常实用，日常web应用通常分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也非常麻烦。用flume的这种组合方式能很好的解决这一问题，每台服务器部署一个flume采集日志，传送到一个集中收集日志的 flume，再由此flume上传到hdfs、hive、hbase等进行日志分析（体现了Flume分布式的思想）。

2.4 Flume 企业开发案例

需求：

使用Flume采集服务器本地日志（此处用端口数据模拟日志），需要按照日志类型的不同，将不同种类的日志发往不同的分析系统。

分析：

在实际的开发中，一台服务器产生的日志类型可能有很多种，不同类型的日志可能需要发送到不同的分析系统。此时会用到Flume拓扑结构中的Multiplexing结构，Multiplexing的原理是，根据event中Header的某个key的值，将不同的event发送到不同的Channel中，所以我们需要自定义一个Interceptor，为不同类型的event的Header中的key赋予不同的值。
在该案例中，我们以端口数据模拟日志，以是否包含“atguigu”模拟不同类型的日志，我们需要自定义interceptor区分数据中是否包含“atguigu”，将其分别发往不同的Channel。

在这里插入图片描述

实现步骤：

（1）创建一个maven项目，并引入以下依赖。

<dependency> 
<groupId>org.apache.flume</groupId> 
<artifactId>flume-ng-core</artifactId> 
<version>1.9.0</version> 
</dependency>

（2）定义CustomInterceptor 类并实现Interceptor 接口。

package com.atguigu.interceptor; 
import org.apache.flume.Context; 
import org.apache.flume.Event; 
import org.apache.flume.interceptor.Interceptor; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Map; 
public class TypeInterceptor implements Interceptor { 
//声明一个存放事件的集合 
private List<Event> addHeaderEvents; 
@Override 
    public void initialize() { 
        //初始化存放事件的集合 
        addHeaderEvents = new ArrayList<>(); 
    } 
 
    //单个事件拦截 
    @Override 
    public Event intercept(Event event) { 
        //1.获取事件中的头信息 
        Map<String, String> headers = event.getHeaders(); 
        //2.获取事件中的body信息 
        String body = new String(event.getBody()); 
        //3.根据body中是否有"atguigu"来决定添加怎样的头信息 
        if (body.contains("atguigu")) { 
            //4.添加头信息 
            headers.put("type", "first"); 
        } else { 
            //4.添加头信息 
            headers.put("type", "second"); 
        } 
 
        return event; 
    } 
 
    //批量事件拦截 
    @Override 
    public List<Event> intercept(List<Event> events) { 
        //1.清空集合 
        addHeaderEvents.clear(); 
        //2.遍历events 
        for (Event event : events) { 
            //3.给每一个事件添加头信息 
            addHeaderEvents.add(intercept(event)); 
        } 
        //4.返回结果 
        return addHeaderEvents; 
    } 
 
    @Override 
    public void close() { 
    } 
 
    public static class Builder implements Interceptor.Builder { 
        @Override 
        public Interceptor build() { 
            return new TypeInterceptor(); 
        } 
        @Override 
        public void configure(Context context) { 
        } 
    } 
 
}

（3）编辑flume配置文件
为hadoop102上的Flume1配置1个netcat source，1个sink group（2个avro sink），并配置相应的ChannelSelector和interceptor。

# Name the components on this agent 
a1.sources = r1 
a1.sinks = k1 k2 
a1.channels = c1 c2 
 
# Describe/configure the source 
a1.sources.r1.type = netcat 
a1.sources.r1.bind = localhost 
a1.sources.r1.port = 44444 
a1.sources.r1.interceptors = i1 
a1.sources.r1.interceptors.i1.type = 
com.atguigu.flume.interceptor.CustomInterceptor$Builder 
a1.sources.r1.selector.type = multiplexing 
a1.sources.r1.selector.header = type 
a1.sources.r1.selector.mapping.first = c1 
a1.sources.r1.selector.mapping.second = c2 
# Describe the sink 
a1.sinks.k1.type = avro 
a1.sinks.k1.hostname = hadoop103 
a1.sinks.k1.port = 4141 
 
a1.sinks.k2.type=avro 
a1.sinks.k2.hostname = hadoop104 
a1.sinks.k2.port = 4242 
 
# Use a channel which buffers events in memory 
a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 
 
# Use a channel which buffers events in memory 
a1.channels.c2.type = memory 
a1.channels.c2.capacity = 1000 
a1.channels.c2.transactionCapacity = 100 
 
# Bind the source and sink to the channel 
a1.sources.r1.channels = c1 c2 
a1.sinks.k1.channel = c1 
a1.sinks.k2.channel = c2

为hadoop103上的Flume4配置一个avro source和一个logger sink。

a1.sources = r1 
a1.sinks = k1 
a1.channels = c1 
 
a1.sources.r1.type = avro 
a1.sources.r1.bind = hadoop103 
a1.sources.r1.port = 4141 
a1.sinks.k1.type = logger 

a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 

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

为hadoop104 上的Flume3配置一个avro source和一个logger sink。

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

a1.sources.r1.type = avro 
a1.sources.r1.bind = hadoop104 
a1.sources.r1.port = 4242 

a1.sinks.k1.type = logger 

a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 

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

（4）分别在hadoop102，hadoop103，hadoop104上启动flume进程，注意先后顺序。
（5）在hadoop102使用netcat向localhost:44444 发送字母和数字。
（6）观察hadoop103和hadoop104打印的日志，发现带有atguigu的数据均被发到了Hadoop103上，其余数据均被发到了Hadoop104。

3. Flume生产经验（重点）

3.1 Flume 参数调优

1）Source

增加Source个数（使用Tair Dir Source时可增加FileGroups个数）可以增大Source的读取数据的能力。
例如：当某一个目录产生的文件过多时需要将这个文件目录拆分成多个文件目录，同时配置好多个Source以保证Source有足够的能力获取到新产生的数据。
参数：
batchSize 参数决定 Source 一次批量运输到Channel 的event 条数，适当调大这个参数可以提高Source搬运Event到Channel时的性能。
2）Channel

type 选择 Memory 时 Channel 的性能最好，但是如果Flume进程意外挂掉可能会丢失数据。type选择File时Channel的容错性更好，但是性能上会比Memory Channel差。使用file Channel 时dataDirs配置多个不同盘下的目录可以提高性能。
参数：
Capacity参数决定Channel可容纳最大的event条数。transactionCapacity 参数决定每次Source 往channel 里面写的最大event 条数和每次Sink 从channel 里面读的最大event 条数。transactionCapacity 需要大于Source 和Sink 的batchSize 参数。
3）Sink

增加Sink 的个数可以增加Sink消费event的能力。Sink也不是越多越好够用就行，过多的Sink会占用系统资源，造成系统资源不必要的浪费。
参数：
batchSize 参数决定 Sink 一次批量从Channel 读取的 event 条数，适当调大这个参数可以提高Sink从Channel搬出event的性能。

3.2 Flume 采集数据会丢失吗?

会，虽然Flume Agent内部有完善的事务机制，Source 到 Channel 是事务性的，Channel到Sink是事务性的，但是在Source到Channel这个过程依然会丢失数据；
例如：使用Flume监听某一端口，当Channel数据满了之后，虽然此时存在事务操作，未提交的数据会doRollback，但是源码中对putList的操作是直接清空，所以存在丢数的情况。还有一钟可能丢失数据的情况是Channel采用MemoryChannel，agent宕机导致数据丢失。
Flume 还有可能造成数据的重复，例如数据已经成功由Sink发出，但是没有接收到响应，Sink会再次发送数据，此时可能会导致数据的重复。

3.3 Flume与Kafka

Kafka 是一个分布式消息中间件，自带存储，提供 push 和 pull 存取数据的功能，是一个非常通用的消息缓存系统，可以有许多生产者和很多的消费者共享多个主题。Kafka 以其高吞吐量、低延迟和可扩展性而适用于实时数据流处理和日志聚合。

Flume 和 Kafka 可以集成使用，以完成实时流式的日志处理。一般使用 Flume + Kafka 来完成这一任务，后面再连接上 Flink/Spark Streaming 等流式实时处理技术，从而完成日志实时解析的目标。如果 Flume 直接对接实时计算框架，当数据采集速度大于数据处理速度时，很容易发生数据堆积或者数据丢失，而 Kafka 可以作为一个消息缓存队列，实现数据的多分发。

通常，在实际应用中，Flume 可以配置为从各种源头采集数据，并将数据发送到 Kafka 中。在 Kafka 中，数据可以进行实时消费，最终实现数据的实时清洗和处理。这种组合可以有效地处理大量的实时数据，并保证数据的可靠性和实时性

例如：在一个电商平台的实时推荐系统中，用户对商品进行评分时，后台可以实时获取这些评分数据。Flume 可以监听日志文件，将评分信息通过 Kafka 发送到下游清洗服务（如SQL或JAVA程序），清洗服务对接收到的数据进行清洗和处理，生成推荐结果，再通过 API 接口返回给前端展示给用户（如报表、看板等）。