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
