Flume架构与实践

Flume是一款在线数据采集的系统，典型的应用场景是作为数据的总线，在线的进行日志的采集、分发与存储，它是一个微内核，可扩展的架构设计，典型的部署图如下：

Flume与Kafka的比较

1、Kafka是pull based，Flume是Push Based， 如果你有很多下游的Data Consumer消费同一份数据，需要做限流、降级等个性化的实现，个人建议用Kafka。

2、Kafka有Replication，Flume配置起来很复杂，如果要求很高的容错性(Data High Availability)，个人建议用Kafka。

3、若果需要更多的一站式Sink实现，例如HDFS，HBase Sink 等，个人建议用Flume。

4、个人推荐的架构Agent—Kafka—flume。

Flume基本概念

Event

Event是flume端到端传输的基本单元，它由数据本身和一些头域信息构成，头域信息的开销对整个数据采集来说是透明的，由K-V构成的Map信息，主要用于上下信息的传递。

Client

它主要用于产生Events，并将它们发送到一个或多个Agents，常见的有Flume log4j Appender、LoadBalancingRpcClient、FailoverRpcClient，它不是部署图中的必须部分。

Source

它是数据源，从特定渠道接受events，将它存储到Channel中，常见Source有如下分类

Agent-to-Agent的Source

Avro Source、ThriftSource

自产生Event的Source

Exec Source、SpoolingDirectory Source

与知名系统对接的Source

Kafka Source、Syslog Sources

Channel

它缓存接收到的events直到它们被Sinks节点消费完成，不同的Channel提供不同持久化层级：

Memory Channel:易失，重启丢数据。

File Channel:以WAL文件为支持，保证数据的本地持久化，重启数据不丢失。

JDBC Channel:以特定的数据库为备份。

所有的通道的存取操作都有“事务”机制的保证，对顺序性是一种弱保护的机制。可以同时对接任意个数的Sources、Sinks。它解耦了upstream与downstream的依赖关系，是整个Flow可靠性、可用性的关键所在。

Sink

它主要从特定的Channel中获取数据，将数据可靠的传输到下一个目的地，保证At least once的消费，一个Sink有且只有一个Channel；常见Sink有

Terminal sink

HDFS Sink、File RollSink、ElasticSearchSink、KafkaSink

Agent-to-Agent Sink

Avro Sink、ThriftSink

Agent

它是一个拥有Sources、Channels、Sinks、将events对象进行透明传输的容器进程。它是Flume数据流的基础组件，提供了生命周期管理、配置动态生效、数据流监控等基本功能。

Configuration

一个配置文件可以同时配置多个Agent，只有指定agent名称相关的配置才会被加载，配置错误的组件在加载的时候输出一个条告警日志，然后会被忽略，Agent支持配置的动态加载。

# Name the components on this agent

a3.sources = r3

a3.sinks = k3

a3.channels = c3

#Describe/configure the source

a3.sources.r3.type= avro

a3.sources.r3.bind=0.0.0.0

a3.sources.r3.port=28080

# Describe the sink

a3.sinks.k3.type =hdfs

a3.sinks.k3.hdfs.path= hdfs://10.0.53.81:9000/%{savePath}/%Y-%m-%d

a3.sinks.k3.hdfs.callTimeout=300000

a3.sinks.k3.hdfs.filePrefix=log

a3.sinks.k3.hdfs.rollInterval=3600

a3.sinks.k3.hdfs.rollSize=1073741824

a3.sinks.k3.hdfs.hdfs.batchSize=4000

a3.sinks.k3.hdfs.threadsPoolSize=30

a3.sinks.k3.hdfs.rollTimerPoolSize=3

a3.sinks.k3.hdfs.rollCount=0

a3.sinks.k3.hdfs.fileType= DataStream

a3.sinks.k3.hdfs.writeFormat= Text

a3.sinks.k3.serializer=text

a3.sinks.k3.serializer.appendNewline=false

a3.sinks.k3.hdfs.minBlockReplicas=1

# Use a channel which buffers events in memory

a3.channels.c3.type=file

a3.channels.c3.checkpointDir=/data/flume/channel/filechannel_3/checkpointDir

a3.channels.c3.dataDirs=/data/flume/channel/filechanne1_3/dataDirs

a3.channels.c3.keep-alive= 10

a3.channels.c3.transactionCapacity=10000

a3.channels.c3.capacity=104857600

# Bind the source and sink to the channel

a3.sources.r3.channels= c3 c

a3.sinks.k3.channel= c3

Flume基本原理

总体介绍

1、客户端发送events到Agents。

2、Agent拥有Source, Interceptors, Channel Selectors, Channels, Sink Processors and Sinks等组件。Source接受Events,经过Interceptor(s)过滤,根据Channel Selector将它们放置到channel(s)中，Sink Processor选择一个Sink从指定的Channel获取Events发送到配置的下一跳节点。

3、Channel的存、取操作是通过“事务的方式”去保证了单节点的消息投递的可靠性；Channel持久化保证了端到端的可靠性。

Flume-Sources

Sources的基本功能

1、从外部的Client或者内部的Sink获取events。

2、将接收到的events存到配置的channel(s)，保证操作的“事务性”。

3、数据分流到不同的目的地

Sources的可用性

1、channel侧put数据的事务保证

2、可靠外部客户端的的异常重试（Avro

sink、LoadBalancingRpcClient）

Flume-Channels

Channels的基本功能

1、作为sources/sinks间的Buffer，能有效抵挡下游消费者的短暂的不可用，下游消费者的消费速度的不匹配。

2、解耦了sources、sinks。

3、多个sources/sinks可以共享同一个channel。

Channels的事务

与DB的事务不是一个概念，它的目标就是保证消息的At Least Once消费，事务是Channel强相关的，它保证了Events一旦“committed”到一个channel，它只有在下游消费者消费并返回了committed.”才会从队列中移除，基本流程如下：

1、Source 1产生的Event, “put” and “committed”到Channel 1。

2、Sink 1从Channel 1中获取并发送到Source 2。

3、Source 2 “puts” and “commits”到Channel 2。

4、Source 2发送成功到Sink 1. Sink 1发送commits确认已“take“成功，将这个event从channel1中删除。

这样就保证了“一批数据的操作的事务性“

Memory Channel

events存储在堆中，存取速度非常快，但是系统Crash、或者进程退出时，存在events丢失。

File Channel

events持久化到本地文件系统中，存取速度相对较慢，但是可靠性高，基本流程如下：

1、events以指定大小存储在log files，持久化到磁盘中

2、指向event的指针（logID+offset）存放在内存队列中

3、queue当前的状态就是events的当前存储状态

4、queue会被周期性的同步到磁盘中- checkpoint

5、channel重启时checkpoint-mmap-ed

6、从上次checkpoint发生的put/take/commit/rollback通过日志回放恢复

Flume-Sinks

它主要从特定的Channel中获取数据，将数据可靠的传输到下一个目的地，保证At least once的消费

HDFS Sink

a3.sinks.k4.type =hdfs

a3.sinks.k4.hdfs.path= hdfs://10.0.53.81:9000/%{savePath}/%Y-%m-%d/%H

a3.sinks.k4.hdfs.callTimeout=300000

a3.sinks.k4.hdfs.filePrefix=log

a3.sinks.k4.hdfs.rollInterval=600

a3.sinks.k4.hdfs.rollSize=1073741824

a3.sinks.k4.hdfs.hdfs.batchSize=4000

a3.sinks.k4.hdfs.threadsPoolSize=30

a3.sinks.k4.hdfs.rollTimerPoolSize=3

a3.sinks.k4.hdfs.rollCount=0

a3.sinks.k4.hdfs.fileType= DataStream

a3.sinks.k4.hdfs.writeFormat= Text

a3.sinks.k4.serializer=text

a3.sinks.k4.serializer.appendNewline=false

a3.sinks.k4.hdfs.minBlockReplicas=1

Contextual Routing

通过Interceptors与Channel Selectors来达到Event的路由功能，例如Terminal Sinks可以直接使用头域信息进行目标Sink节点的选择

Interceptor

它主要用来对流入Source的Event对象进行修饰和过滤，当前内置的Interceptors支持添加时间戳、主机、静态标签等头域信息；支持用户自定义的Interceptor。

Channel Selector

它允许Source根据Event头域的标签信息，从所有的Channels中选出Event对应的Channel；目前内置的Channel Selectors有：

Replicating:将Event应用到所有对应Channel中。

Multiplexing:根据头域选择对应的Channel。

Flume可用性与可靠性

Flume可用性设计

Client可用性

LoadBalancingRpcClient、FailoverRpcClient，保证了Client侧的高可用性。

Channel可用性

Memory Channel、FileChannel、JDBC Channel的可用性递增，性能递减。

Sink Processor

它负责从指定的SinkGroup中调用一个Sink节点，由Sink Runner负责调度，类似做了一层代理；一个Sink节点只能存在于一个SinkProcessor中，如果没有配置任何SinkGroup，默认是在Default Sink Processor中。

目前内置Sink Process有 Load Balancing Sink Processor--RANDOM,ROUND_ROBIN 、Failover Sink Processor、Default Sink Processor

Flume可靠性设计

channel的事务

保证了Agent间数据交换的At least once消费 

内置的负载均衡和FailOve机制

channel的持久化

保证了Agent挂了之后的自动恢复，保证了消息的At least once消费

节点不可用

1、通过冗余的topologies来解决，Flume允许你将你数据流复制到冗余的topologies，这个对于应对磁盘损坏和单点失效有用，代价较大，topology会比较复杂。

2、如果承载channel的磁盘做了Raid，重新挂在，启动agent，也能很快的进行数据恢复，例如Kafka channel/JDBC Channel

3、允许部分数据的丢失。

4、概率较小，推荐在应用层进行回放的操作。

Flume的调优

选择正确的Channel

memory channel：低延时，高吞吐，可靠性弱。

file channel：延时相对高，吞吐相对小，可靠性较高。

disk-based channels10’s of MB/s

memory based channels100’s of MB/s

选择合适的capacity

根据你容忍的down机时间而定，capacity越大，恢复需要的时间越长，同时存在消费延时的问题。

选择合适的batch size

batch size越大吞吐量越高，异常重传的代价高，一般推荐100 or 1,000 or 10,000这个跟单个event的大小也有关

client与agent的比例配比

20:1 or 30:1的客户端agent比例比较适中

GC参数的选择

根据应用的具体情况进行调优

agent的监控

通过agent暴露的metrice服务，关注ChannelSize找到数据流中的瓶颈

http://10.0.52.28:34545/metrics

Flume应用

业界应用

我司应用

使用的是Client+Avro Source+ Replicating/ Multiplexing+TimeStampInterceptor+FileChannel/MemChannel+Failover sink Processr/Loadbalance SinkProcess+HdfsSink/ ElasticSink/FileSink/CustomSource