Flume

kidys

简介


　　Flume是一个分布式，可靠的，可以用来有效的收集，聚合和移动大的日志数据的可用服务。它是一个基于数据流的简单且灵活的框架。采用可调可靠性机制和多种故障转移及恢复机制保证很好的稳健性和容错性。它使用了一个简单的可扩展的数据模型，允许在线分析应用。data flow（数据流）是Flume最重要的抽象，data flow描述了从数据产生，传输、处理并最终写入目标路径。下图描述了Flume重要的几个组件：

　　结合上图，Flume的一些核心组件

组件	功能
Web Server	数据产生的源头。
Agent	Flume的核心就是Agent 。Agent是一个Java进程，且运行在日志收集端，通过Agent接收日志，然后暂存起来，再发送到目的地。
Source	Agent核心组件之一，Source（源）用于从Web Server收集数据，然后发送到Channel（通道）。
Channel	Agent核心组件之一，Channel（通道）可以用来从Source接收数据，然后发送到Sink，Channel存放零时数据，有点类似队列一样。
Sink	Agent核心组件之一，Sink（接收器）用来把数据发送的目标地点，如上图放到HDFS中。
Event	整个数据传输过程中，流动的对象都是实现了org.apache.flume.Event接口的对象。Event也是事务保证的级别。
Flow	Event从源点到达目的点的迁移的抽象

注：Flume支持多个Agent，支持扇入(fan-in)、扇出(fan-out)。




系统要求
　　1，java运行时环境--Java 1.6或者更晚的（建议Java 1.7）
　　2，内存 - 使用Sources，Channels或者Sinks要配置足够的内存。
　　3，硬盘空间 - 使用Channels和Sinks要配置足够的硬盘空间。
　　4，目录权限 - 被Agent使用的目录必须要有读和写的权限。
　　官方网站：http://flume.apache.org/

用户文档：http://flume.apache.org/FlumeUserGuide.html

开发文档：http://flume.apache.org/FlumeDeveloperGuide.html


数据流模型


　　Flume Event被定义为一个字节的有效载荷以及以及一个可选的字符串属性集。Flume Source接收来自外部源的数据（比如：Web Server）。Flume的数据流由Event贯穿始终。事件是Flume的基本数据单元，当Source捕获事件后会进行特定的格式化，然后Source会把Event推送到一个或多个Channel中，可以把Channel看作是一个缓冲区或者一个队列，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。
　　Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source,Channel和Sink可以自由组合。也可以自定义Source，Channel和Sink。关于这三者的组合关系是很灵活的，下面我们看下几幅图：
　　多级代理：
　　multi-hop模式：事件在到达最终目的地之前通过多个Agent

　　合并：

　　Multiplexing：
　　Flume支持multiplexing Event到一个或者多个目的地，它通过多路复用器将Event复制或者选择性的路由到一个或多个Channel。

　　上图的示例，名称为“foo”的Agent，Source会Fan out到三个Channel中，Fan out的方式分为replicating和multiplexing。如果采用replicating方式，那么每个Event会发到这三个Channel中，如果采用multiplexing方式，它是根据Event的属性来路由，然后发送到指定的Channel中。
　　注：一个sink只能接收一个Channel发过来的数据

　　下面我们分别举两个例子：
　　replicating：


# List the sources, sinks and channels for the agent
<Agent>.sources = <Source1>
<Agent>.sinks = <Sink1> <Sink2>
<Agent>.channels = <Channel1> <Channel2>
# set list of channels for source (separated by space)
<Agent>.sources.<Source1>.channels = <Channel1> <Channel2>
# set channel for sinks
<Agent>.sinks.<Sink1>.channel = <Channel1>
<Agent>.sinks.<Sink2>.channel = <Channel2>
<Agent>.sources.<Source1>.selector.type = replicatingreplicating是将Event发送到所有的Channel。
　　multiplexing：

# Mapping for multiplexing selector
<Agent>.sources.<Source1>.selector.type = multiplexing
<Agent>.sources.<Source1>.selector.header = <someHeader>
<Agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1>
<Agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2>
<Agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2>
#...
<Agent>.sources.<Source1>.selector.default = <Channel2>
　　multiplexing是根据header的值来路由的，根据上面的例子，如果Event header的<someHeader>属性值为Value1的话，就发送Channel1，如果属性值为Value2的话就发送到Channel1,和Channel2，如果属性值为Value3的话，就发送到Channel2中，其他的都发送到Channel2.
　　


可靠性
　　Event是分阶梯传送的，从Source->Channel->Sink的，Event最终传送到下一个Agent或下一个终端资源库（例如HDFS，文件系统等），在传送打到后，Event在Channel中删除。
　　end-to-end指的是：在删除Channel中的Event时，保证Event已经传递到了下一个Agent或者终端资源库，但是这里没有提到Source进入到Channel之前如果保证不丢失。具体的可以看下每个Source的实现。
　　Flume采用了事务的方式来保证Event的可靠性传输，保证Event集合在点对点的传输是可靠的。

　　


可恢复性
　　事件在Channel是分阶梯，管理从失败中恢复。Flume支持持久化到本地文件系统中（即：FileChannel，但是性能不佳）。Memory Channel只是存储Event到内存的队列中，性能很好，但是当Agent挂掉，内存中的Event是不能够被恢复的。

　　


安装第三方插件
　　Flume是完全基于插件的框架，Flume有多种Sources，Channels，Sinks，Serializers。
　　虽然可以包括自定义Flume组件，但是加入他们所需的jars到conf/flume-env.sh文件中FLUME_CLASSPATH变量中，Flume现在支持一种特殊的目录结构叫做：plugins.d，它有特殊的格式，这样的话Flume可以自动的启用插件。这样就可以很容易的管理插件包装的问题，也方便调试和排除故障，特别是Lib包冲突的问题。

plugins.d目录
　　plugins.d目录位于$FLUME_HOME/plugins.d。在启动的时候，flume-ng启动脚本会检查 plugins.d 目录的插件确保符合下面的格式，并且包含了正确的路径。

插件目录布局
　　每个在 plugins.d 内的插件，最多包含三个子目录。
　　1，lib - 插件的JAR。
　　2，libext - 插件依赖JAR（S）
　　3，native - 任何所需的本地库，例如：.so文件

下面是两个插件在 plugins.d 目录中的位置：
plugins.d/
plugins.d/custom-source-1/
plugins.d/custom-source-1/lib/my-source.jar
plugins.d/custom-source-1/libext/spring-core-2.5.6.jar
plugins.d/custom-source-2/
plugins.d/custom-source-2/lib/custom.jar
plugins.d/custom-source-2/native/gettext.so


基于Zookeeper的配置
　　Flume支持Agent的配置信息通过Zookeeper，但是这个是一个实验性的功能，不建议在生产环境使用。这个配置文件需要上传到Zookeeper中，这个配置文件将储存在Zookeeper的节点下，下面看Agent a1和a2在zookeeper的结构：

- /flume
|- /a1 [Agent config file]
|- /a2 [Agent config file]配置文件上传后，使用如下参数启动Agent：
$ bin/flume-ng agent –conf conf -z zkhost:2181,zkhost1:2181 -p /flume –name a1 -Dflume.root.logger=INFO,console
参数名称
默认
描述


z


–

Zookeeper服务器列表，格式是hostname:port，用逗号分隔



p


/flume

Zookeeper存储Agent配置文件的路径。
　　详情查看：Flume采用zookeeper管理配置


　　


Flume Sources

名称	描述
Avro Source	监听Avro端口并且从外部Avro客户端接收Event。
Thrift Source	监听Thrift端口并且从外部Thrift客户端接收Event。
Exec Source	运行Unix命令，预计能产生持续的输出（）
JMS Source	从JMS的queue或者topic读取数据
Spooling Directory Source	监测配置的目录下新增的文件，并将文件中的数据读取出来，可实现准实时。
Kafka Source	作为一个消费者，从Apache Kafka的topic读取消息。
NetCat Source	监听指定端口，将每一行封装成一个Event。
Sequence Generator Source	一个简单的序列生成器，从0开始，每次增加1，主要用于测试。
Syslog TCP Source	读取syslog消息并且生成Flume的Event，TCP每个Event用'n'分割
Multiport Syslog TCP Source	这个是最新，最快，多端口版本的Syslog TCP Source。
Syslog UDP Source	读取syslog消息并且生成Flume的Event，每个输入信息当成一个Event
HTTP Source	接受HTTP的GET或者POST数据，GET是实验性质的，不建议使用。支持JSON、BLOB表示形式
Stress Source	是一个内部负载生成源，用来做压力测试是非常有用的
Avro Legacy Source	允许Flume 1.x的Agent从Flume 0.9.4的Agent中获取数据，它接受0.9.4的Event，并且转化成1.x
Thrift Legacy Source	和Avro Legacy Source雷同
Scribe Source

Flume Sinks

名称	描述
HDFS Sink	将数据写入到HDFS
Hive Sink	将文本或者JSON数据用分隔符分割，直接变成Hive的表，或者是分区
Logger Sink	记录Event的Info级别日志，通常用于测试或调试。
Avro Sink	采用Avro Sink接收到的Event，发送到另外一个Avro Source
Thrift Sink	采用Thrift Sink接收到的Event，发送到另外一个Thrift Source
IRC Sink	从附加的Channel获取数据，转发到配置中的IRC的目的地。IRC（类似于网络聊天室的服务）
File Roll Sink	将Event存放到本地文件系统，根据时间或者大小生成文件。
Null Sink	丢弃所有从Channel获取的Event。
HBaseSink	写入数据到Hbase
AsyncHBaseSink	采用异步的形式写入数据到Hbase
MorphlineSolrSink	将数据写入Solr集群
ElasticSearchSink	将数据写入ElasticSearch集群
Kafka Sink	将数据写入Kafka集群

Flume Channels

名称	描述
Memory Channel	储存Event在内存队列中，如果宕机可能会造成数据的丢失，具有很高的吞吐量。
JDBC Channel	将Event储存在持久化的数据库中，目前支持嵌入式的Derby。
Kafka Channel	将Event储存在Kafka集群（必须单独部署），Kafka提供了高可用和复制性，所以Kafka或者Agent崩溃，数据也不会丢失。
File Channel	将Event持久化在本地文件系统里(性能较差)，但是可以保证数据不丢失。
Spillable Memory Channel	Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件（当前试验性的，不建议生产环境使用）
Pseudo Transaction Channel	仅用于单元测试，不能用于生产环境。