Flume事务（Transaction）

解释：
事务的数据回滚：
当数据没有执行成功，需要回滚到原来的状态，保持数据的一致性、原子性
防止数据执行到一半让数据产生异步变化。

!!!batchSize<=transactionCapacity<=capacity

Flume Agent 内部原理

重要组件：

1）ChannelSelector（步骤4）
ChannelSelector的作用就是选出Event将要被发往哪个Channel。其共有两种类型，分别是Replicating（复制）和Multiplexing（多路复用）。
ReplicatingSelector：将同一个Event发往所有的Channel，（一对多）
Multiplexing：根据相应的原则，将不同的Event发往不同的Channel（多对多（或多对1，因为可以通过配置设定））

2）SinkProcessor（步骤7）
SinkProcessor共有三种类型，分别是DefaultSinkProcessor、LoadBalancingSinkProcessor和FailoverSinkProcessor
DefaultSinkProcessor对应的是单个的Sink，
LoadBalancingSinkProcessor和FailoverSinkProcessor对应的是Sink Group，
LoadBalancingSinkProcessor可以实现负载均衡的功能，
FailoverSinkProcessor可以错误恢复的功能。
两者一般配合使用

3）Interceptor
通过拦截器给数据赋上header

Flume拓扑结构(Avro)

简单串联

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

复制和多路复用

![image.png](https://cdn.nlark.com/yuque/0/2021/png/21361442/1619610882356-d0214a31-4ea2-4e45-be84-4f076f06d5f8.png#crop=0&crop=0&crop=1&crop=1&height=283&id=TpVg4&margin=%5Bobject%20Object%5D&name=image.png&originHeight=283&originWidth=472&originalType=binary&ratio=1&rotation=0&showTitle=false&size=48941&status=done&style=none&title=&width=472)<br />Flume支持将事件流向一个或者多个目的地。（channel selector决定）<br />这种模式可以将相同数据复制到多个channel中，（ReplicatingSelector）<br />或者将不同数据分发到不同的channel中，sink可以选择传送到不同的目的地。（Multiplexing）

负载均衡和故障转移

Flume支持使用将多个sink逻辑上分到一个sink组，sink组配合不同的SinkProcessor可以实现负载均衡和错误恢复的功能。
意义：
扩大缓冲区大小，提升缓冲能力
当一条坏了，另外两台可以开启用以备用

聚合

这种模式是我们最常见的，也非常实用，日常web应用通常分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也非常麻烦。用flume的这种组合方式能很好的解决这一问题，每台服务器部署一个flume采集日志，传送到一个集中收集日志的flume，再由此flume上传到hdfs、hive、hbase等，进行日志分析。
如果数据量不大，产生很多小文件，聚合可以方便统一提交，类似于缓冲区的功能

企业开发案例

1、复制和多路
2、负载均衡和故障转移
3、聚合
具体看文档第三章3.4节
搜索“开发案例”
尚硅谷大数据技术之Flume.docx

自定义Interceptor

1）案例需求
使用Flume采集服务器本地日志，需要按照日志类型的不同，将不同种类的日志发往不同的分析系统。

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

3）实现步骤
（1）创建一个maven项目，并引入以下依赖。

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

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


import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;

public class CustomInterceptor implements Interceptor {


    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        byte[] body = event.getBody();
        if (body[0] < 'z' && body[0] > 'a') {
            event.getHeaders().put("type", "letter");
        } else if (body[0] > '0' && body[0] < '9') {
            event.getHeaders().put("type", "number");
        }
        return event;

    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {

    }
    //静态内部类构建本对象CustomInterceptor
    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new CustomInterceptor();
        }

        @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.interceptor.CustomInterceptor$Builder
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.letter = c1
a1.sources.r1.selector.mapping.number = c2
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141

a1.sinks.k2.type=avro
a1.sinks.k2.hostname = hadoop102
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 = hadoop102
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 = hadoop102
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打印的日志。

自定义Source

flume视频day2

自定义Sink