Flume-NG配置说明文档V1.0版

1、Flume-NG配置说明文档文件状态： 草稿 正式发布 正在修改文件标识：说明文档当前版本：V1.0作 者：审 核：完成日期：Flume介绍Flume是Cloudera提供的日志收集系统，后贡献给了Apache。Flume支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方(可定制)的能力。Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统，可有效地收集，汇总和来自许多不同来源的大量日志数据到集中的数据存储系统。Apache的Flume是在Apache软件基金会的顶级项目。目前有两个版本的代码行，0.9.x版本(也称

2、为Flume-OG Flumeoriginal generation)版本以及1.x的版本(也称为Flume-NG Flume next generation版本)。Flume NG 1.x 是Flume 0.9.x的重构版本，去掉了master、zookeeper、collector以及Web console，只有source、sink、channel，成为一个数据传输工具。Flume-NG支持多种source、多种channel、多种sink，而且可扩展性好，可以自定义组件，是一个非常优秀的日志传输系统。FlumeNG数据获取Flume提供了各种source的实现，包括Avro Sourc

3、e、Exce Source、Spooling Directory Source、NetCat Source、Syslog Source、Syslog TCP Source、Syslog UDP Source、HTTP Source、HDFS Source，etc。RPC1、在flume中 ，Avro客户端使用AVRO RPC机制可以发送一个给定的文件 Avro 源：2、$ bin/flume-ng avro-client -H localhost -p 41414 F /usr/logs/log.103、上面的命令将发送的/ usr/logs/log.10的内容到 flume源监听端Execu

4、ting commands1、还有一个exec执行一个给定的命令获得输出的源。一个单一的输出，即“line”。回车（ R）或换行符（ N），或两者一起的文本。注：Flume不支持tail做为一个源，不过可以通过exec tail。Network streamsFlume支持以下的机制，从流行的日志流类型读取数据1、Avro(数据序列化系统)2、Syslog3、Netcat(使用TCP或UDP协议的网络连接去读写数据)Flume部署种类Flume可以实现多代理、合并、多路复用等多种部署方式。多代理：设置一个多层的流，你需要有一个指向下一跳avro源的第一跳的avro 接收器。这将导致第一Flum

5、e代理转发事件到下一个Flume代理。例如，如果您定期发送的文件，每个事件（1文件）AVRO客户端使用本地Flume 代理，那么这个当地的代理可以转发到另一个有存储的代理。如下图：合并：在日志收集的一个非常普遍的情况是大量生产客户日志的数据发送到一些消费者代理连接到存储子系统。举例来说，从数以百计的Web服务器收集的日志发送到十几代理写入HDFS集群。如下图：多路复用流：Flume支持从一个源到多个通道。有两种模式的，分别是复制和复用。在复制的情况下，流的事件被发送到所有的配置通道。在复用的情况下，事件被发送到可用的渠道中的一个子集。如下图：Flume-NG配置Flume-NG安装简单，使用方

6、便。从网上下载flume压缩包解压，配置其中的配置文件，即可使用。定义流Flume事件(event)被定义为一个单位的数据流量有一个字节的有效载荷和一个可选字符串属性。Flume代理是一个承载组件（source源，sink接收器或channel通道）的(JVM)进程，通过事件流从外部源到下一个目的地（跳）。一个web服务器的产生的事件由 Flume源消耗。外部源发送事件发送到Flume中，会带着一个识别的格式。例如: 例如:一个Avro Flume源可以用来接收从Avro clients或其他flume代理从Avro link发送事件。当一个Flume 源接收一个事件，他会存储到一个或多个ch

7、annel中，这些channel会一直保存着event，直到被Flume sink处理掉，例如JDBC Channel作为一个例子-它使用一个文件系统支持嵌入式数据库，sink从channel中移除事件，同时放入到一个外部的仓库，比如HDFS，或者流转到下一个Flume source 源，source和sink在agent中是以异步运行方式运行事件。Flume代理配置存储在本地配置文件。这是一个文本文件格式如Java属性文件格式。在相同的配置文件中，可以指定一个或多个代理的配置。配置一个代理需要先单独配置每个组件（source、sink、channel），然后连接在一起，形成数据流。在conf

8、目录下的flume.conf（无此文件则创建）中配置。在Flume中，最重要的抽象是data flow(数据流)，data flow描述了数据从产生，传输、处理并最终写入目标的一条路径。配置单个代理举一个例子，配置文件，描述一个单节点的Flume部署。这种配置可以让用户生成的事件和随后输出到控制台。将以下代码输入flume.conf配置文件。#确定代理上的组件agent1.sources = source1agent1.sinks = sink1agent1.channels = channel1#配置sourceagent1.sources.source1.type= netcatagent

9、1.sources.source1.bind = localhostagent1.sources.source1.port = 44444#配置sinkagent1.sinks.sink1.type= logger#配置channel（内存类型）agent1.channels.channel1.type= memoryagent1.channels.channel1.capacity = 1000agent1.channels.channel1.transactionCapactiy = 100#将组件连接agent1.sources.source1.channels = channel1ag

10、ent1.sinks.sink1.channel = channel1由以上代码可知，代理的名称为agent1，sink类型为日志，channel类型为内存，source为netcat方式，总体流程为：从netcat获取数据，通过内存输出到日志这个配置定义了一个单一的代理，称为agent1。agent1监听44444端口的数据作为source，通道缓存在内存中事件数据，事件数据记录到控制台和一个接收器上的数据源。配置文件名的各个组成部分，然后介绍了他们的 类型和配置参数。一个给定的配置文件可能会定义多个命名的代理人;一个给定的Flume进程启动时传递一个标志，告诉它的具名代理体现。在flume

11、-ng目录下，结合此配置文件，我们启动Flume按如下参数：1bin/flume-ng agent -f conf/flume.conf -n agent1 c ./conf -Dflume.root.logger=INFO,console其中-n指定代理名称，-c指定配置文件目录，-f指定代理的配置文件。请注意，在完整部署，我们通常会包括一个选项： - CONF=，可用-c代替。 目录将包括一个shell脚本flume-env.sh（没有则创建此文件，其内容可参考flume-env.sh.template）和内置的Log4j属性文件。在这个例子中，我们使用一个Java选项强制flume登录到

12、控制台。我们可以从一个单独的终端，然后telnet端口44444和发送flume事件：telnet localhost 44444Trying .Connected to localhost.Escape character is .1234OK他原来的flume终端输出日志信息的事件。2013-04-17 14:41:53,499 (lifecycleSupervisor-1-3) INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:150) Source starting2013-04-1

13、7 14:41:53,513 (lifecycleSupervisor-1-3) INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:164) Created serverSocket:sun.nio.ch.ServerSocketChannelImpl/:444442013-04-17 14:42:10,898 (SinkRunner-PollingRunner-DefaultSinkProcessor) INFO - org.apache.flume.sink.LoggerS

14、cess(LoggerSink.java:70) Event: headers: body: 31 32 33 34 0D 1234. 说明你已经成功地配置和部署了一个flume代理！合并每个Flume代理均需要对其source、sink、channel配置。source是数据源用以指定数据来源，sink是接收器用以指向数据目标流向，channel是通道用以传输数据。以下以合并为例，讲述配置过程：上图中以tongjitest166和tongjitest167作为数据收集代理weblog-agent，tongjitest165作为数据合并代理hdfs-agent，并将数据存入hdfs中。Flum

15、e所有配置文件在conf下，需要自己创建两个文件：flume-env.sh(可参考flume-env.sh.template)和flume.conf，其中后者名称可以随意指定。flume-env.sh主要配置：JAVA_HOME(指向本地java安装目录)、JAVA_OPTS(设置JVM相关运行参数)、FLUME_CLASSPATH(指向flume配置文件目录)，flume.conf配置source、sink、channel的文件。tongjitest166和tongjitest167配置完全相同(也可以根据需要，配置不同的source端)。一、weblog-agent代理配置此代理配置除了配

16、置source、sink、channel以外。因为是多层代理，所以还要有“一个指向下一跳avro源的第一跳的avro 接收器”。如下，两个weblog-agent的flume.conf配置：# weblog agent config#确定sources, sinks和channels组件weblog-agent.sources = avro-AppSrv-sourceweblog-agent.sinks = avro-forward-sinkweblog-agent.channels = jdbc-channel#定义流weblog-agent.sources.avro-AppSrv-sour

17、ce.channels = jdbc-channelweblog-agent.sinks.avro-forward-sink.channel = jdbc-channel#指向下一跳avro源的第一跳的avro接收器weblog-agent.sources.avro-forward-sink.type = avroweblog-agent.sources.avro-forward-sink.hostname = 65weblog-agent.sources.avro-forward-sink.port = 10000#配置sinkweblog-agent.sinks.avr

18、o-forward-sink.type=avroweblog-agent.sinks.avro-forward-sink.hostname=65weblog-agent.sinks.avro-forward-sink.port=10000#配置channelweblog-agent.channels.jdbc-channel.type= memoryweblog-agent.channels.jdbc-channel.capacity = 1000weblog-agent.channels.jdbc-channel.transactionCapactiy = 1000#配置

19、sourceweblog-agent.sources.avro-AppSrv-source.type= execweblog-agent.sources.avro-AppSrv-mand= tail -f /home/hadoop/t.txtweblog-agent.sources.avro-AppSrv-source.batchSize = 20source、sink、channel都有多种类型，可以根据需要灵活的进行配置。具体参数配置可参考网络资源。二、hdfs-agent代理配置tongjitest165也就是hdfs-agent会将两个weblog-agent的数据进行汇总，并存入hd

20、fs当中。其配置信息如下：# hdfs-agent config#确定sources, sinks和channels组件hdfs-agent.sources = avro-AppSrv-sourcehdfs-agent.sinks = avro-forward-sink hdfs-agent.channels = jdbc-channel#定义流hdfs-agent.sinks.avro-forward-sink.channel = jdbc-channelhdfs-agent.sinks.file-sink.channel=jdbc-channel#配置流向hdfs的sinkhdfs-age

21、nt.sinks.avro-forward-sink.type=hdfshdfs-agent.sinks.avro-forward-sink.hdfs.path= hdfs:/65:8020/flume/webdatahdfs-agent.sinks.avro-forward-sink.hdfs.fileType=DataStreamhdfs-agent.sinks.avro-forward-sink.hdfs.batchSize=1000hdfs-agent.sinks.avro-forward-sink.hdfs.filePrefix=temphdfs-agent.si

22、nks.avro-forward-sink.hdfs.rollCount=2000#配置channelhdfs-agent.channels.jdbc-channel.type= memoryhdfs-agent.channels.jdbc-channel.capacity = 1000hdfs-agent.channels.jdbc-channel.transactionCapactiy = 1000#配置source(指向weblog-agent流向的端口)hdfs-agent.sources.avro-AppSrv-source.type = avrohdfs-agent.sources

23、.avro-AppSrv-source.bind = 65hdfs-agent.sources.avro-AppSrv-source.port = 10000三、启动若是多个代理，需要启动每个代理。如果当前在flume-ng的根目录下，weblog-agent则需要执行bin/flume-ng agent n weblog-agent c ./conf f conf/flume.conf；hdfs-agent则需要执行bin/flume-ng agent n hdfs-agent c ./conf f conf/flume.conf。多路复制当有多个通道，需要将源发送到所有

24、通道时，可以考虑使用复制。将会把源数据复制发送到所有通道。Flume.conf配置信息如下：# weblog agent config#List sources, sinks and channels in the agentweblog-agent.sources = avro-AppSrv-sourceweblog-agent.sinks = avro-forward-sink file-sink log-sinkweblog-agent.channels = jdbc-channel file-channel log-channel#define the flowweblog-agent

25、.sources.avro-AppSrv-source.channels = jdbc-channel file-channel log-channelweblog-agent.sinks.avro-forward-sink.channel = jdbc-channelweblog-agent.sinks.file-sink.channel=file-channelweblog-agent.sinks.log-sink.channel=log-channelweblog-agent.sources.avro-AppSrv-source.selector.type=replicating#avr

26、o sink propertiesweblog-agent.sinks.file-sink.type = file_rollweblog-agent.sinks.file-sink.sink.directory = /home/hadoop/flumeweblog-agent.sinks.file-sink.sink.rollInterval = 0weblog-agent.sinks.log-sink.type=loggerweblog-agent.sinks.avro-forward-sink.type=hdfsweblog-agent.sinks.avro-forward-sink.hd

27、fs.path= hdfs:/65:8020/flume/webdataweblog-agent.sinks.avro-forward-sink.hdfs.fileType=DataStreamweblog-agent.sinks.avro-forward-sink.hdfs.batchSize=20weblog-agent.sinks.avro-forward-sink.hdfs.filePrefix=tempweblog-agent.sinks.avro-forward-sink.hdfs.rollCount=50#configure other piecesweblo

28、g-agent.channels.jdbc-channel.type= memoryweblog-agent.channels.jdbc-channel.capacity = 1000weblog-agent.channels.jdbc-channel.transactionCapactiy = 1000weblog-agent.channels.file-channel.type=memoryweblog-agent.channels.file-channel.capacity = 1000weblog-agent.channels.file-channel.transactionCapac

29、tiy = 1000weblog-agent.channels.log-channel.type=memoryweblog-agent.channels.log-channel.capacity=1000weblog-agent.channels.log-channel.transactionCapactiy=1000#读取本地文件/home/hadoop/t.txtweblog-agent.sources.avro-AppSrv-source.type= execweblog-agent.sources.avro-AppSrv-mand = cat /home/hadoop/t.txtweb

30、log-agent.sources.avro-AppSrv-source.batchSize = 20注：使用复制功能需要“.sources.selector.type = replicating”，如果没有这句，则源数据将会只发送到第一个通道，其余通道在flume启动时会报错；如果一个源通过一个通道发往多个sink，则每个sink的内容只是源的一部分数据，综合所有sink数据才会还原源数据。而且即使有上述复制语句但是只有一个通道，数据同样会分散在多个sink中。启动flume语句：bin/flume-ng agent -f conf/flume.conf -n weblog-agent -c

31、 ./conf -Dflume.root.logger=INFO,console 以上配置代码有一个source组件（使用exec源读取本地文件），三个sink组件（一个写入本地文件、一个写入日志文件、一个写入HDFS）和三个channel组件（都是memory），这些个组件均需要单独配置，并通过sources的channels连接在一起。多路复用多路复用可以根据设定的信息，进一步分流。在使用多路复用前得先设定header，而这在从源读数据时设定。而根据header的值选择推送的channel。Flume.conf配置信息如下：# weblog agent config#List source

32、s, sinks and channels in the agentweblog-agent.sources = avro-AppSrv-sourceweblog-agent.sinks = avro-forward-sink file-sink log-sinkweblog-agent.channels = jdbc-channel file-channel log-channel#define the flowweblog-agent.sources.avro-AppSrv-source.channels = jdbc-channel file-channel log-channelweb

33、log-agent.sinks.avro-forward-sink.channel = jdbc-channelweblog-agent.sinks.file-sink.channel=file-channelweblog-agent.sinks.log-sink.channel=log-channelweblog-agent.sources.avro-AppSrv-source.selector.type=multiplexingweblog-agent.sources.avro-AppSrv-source.selector.header=stateweblog-agent.sources.

34、avro-AppSrv-source.selector.mapping.CA=jdbc-channel log-channelweblog-agent.sources.avro-AppSrv-source.selector.mapping.NY=file-channelweblog-agent.sources.avro-AppSrv-source.selector.mapping.AZ=jdbc-channelweblog-agent.sources.avro-AppSrv-source.selector.default=jdbc-channel#avro sink propertiesweb

35、log-agent.sinks.file-sink.type = FILE_ROLLweblog-agent.sinks.file-sink.sink.directory = /home/hadoop/flumeweblog-agent.sinks.file-sink.sink.rollInterval = 0weblog-agent.sinks.log-sink.type=loggerweblog-agent.sinks.avro-forward-sink.type=hdfsweblog-agent.sinks.avro-forward-sink.hdfs.path= hdfs:/10.32

36、.21.165:8020/flume/webdataweblog-agent.sinks.avro-forward-sink.hdfs.fileType=DataStreamweblog-agent.sinks.avro-forward-sink.hdfs.batchSize=20weblog-agent.sinks.avro-forward-sink.hdfs.filePrefix=tempweblog-agent.sinks.avro-forward-sink.hdfs.rollCount=50#configure other piecesweblog-agent.channels.jdb

37、c-channel.type= memoryweblog-agent.channels.jdbc-channel.capacity = 1000weblog-agent.channels.jdbc-channel.transactionCapactiy = 1000weblog-agent.channels.file-channel.type=memoryweblog-agent.channels.file-channel.capacity = 1000weblog-agent.channels.file-channel.transactionCapactiy = 1000weblog-age

38、nt.channels.log-channel.type=memoryweblog-agent.channels.log-channel.capacity=1000weblog-agent.channels.log-channel.transactionCapactiy=1000weblog-agent.sources.avro-AppSrv-source.type= com.source.ExecSourceweblog-agent.sources.avro-AppSrv-mand = cat /home/hadoop/t.txtweblog-agent.sources.avro-AppSr

39、v-source.batchSize = 20com.source.ExecSource是自定义源，此源是源代码中ExecSource类的修改，在run()方法中eventList.add(EventBuilder.withBody(line.getBytes()改为：eventList.add(EventBuilder.withBody(line.getBytes(),map)其中map是这样声明的：Map map=new HashMap();map.put(state, NY);其中state对应于selector.header，NY对应于mapping.NY，通过这样在从源获取数据时对数

40、据做了封装，成为event，并加入header作为分流标记。当event要推送入channel时，会根据header的值将其推送入相应的channel中。上例中，通过增加header信息将所有数据标记为“NY”，则数据只会推送入file-channel，最终存入/home/hadoop/flume目录下，而其他channel则没有数据。Flume拦截器InterceptorsFlume可以修改和删除传输中的数据。Flume可以通过拦截器拦截符合要的event并设置header，而这可能影响到以后汇总时的复用。例如，可以将不同机器收集的数据加上主机名作为header，然后在汇总端可以按照不同的主

41、机名发送到不同的目标去。举例：多级代理，tongjitest166收集的数据均对其加header为主机名的标记，然后发送到tongjitest165上，然后在复用模式下，将根据header的值选择存储在本地文件还是hdfs中。tongjitest166中的flume配置文件如下：weblog-agent.sources = avro-AppSrv-sourceweblog-agent.sinks = avro-forward-sinkweblog-agent.channels = jdbc-channel#define the flowweblog-agent.sources.avro-App

42、Srv-source.channels = jdbc-channelweblog-agent.sinks.avro-forward-sink.channel = jdbc-channel#avro sink propertiesweblog-agent.sources.avro-forward-sink.type = avroweblog-agent.sources.avro-forward-sink.hostname = 65weblog-agent.sources.avro-forward-sink.port = 10000#weblog-agent.sinks.avr

43、o-forward-sink.type=avroweblog-agent.sinks.avro-forward-sink.hostname=65weblog-agent.sinks.avro-forward-sink.port=10000#weblog-agent.channels.jdbc-channel.type= memoryweblog-agent.channels.jdbc-channel.capacity = 1000weblog-agent.channels.jdbc-channel.transactionCapactiy = 100#weblog-agent

44、.sources.avro-AppSrv-source.type= netcat#weblog-agent.sources.avro-AppSrv-source.bind = localhost#weblog-agent.sources.avro-AppSrv-source.port = 44444#指定源weblog-agent.sources.avro-AppSrv-source.type= execweblog-agent.sources.avro-AppSrv-mand = cat /home/hadoop/t.txtweblog-agent.sources.avro-AppSrv-s

45、ource.batchSize = 100#使用HostInterceptor拦截器weblog-agent.sources.avro-AppSerceptors = i1weblog-agent.sources.avro-AppSerceptors.i1.type= host#不允许使用IP作为主机名weblog-agent.sources.avro-AppSerceptors.i1.useIP=falseweblog-agent.sources.avro-AppSerceptors.i1

46、.preserveExisting = false#设置event的headerweblog-agent.sources.avro-AppSerceptors.i1.hostHeader= hostname 上述配置文件将使用EXEC源读取本地文件，并将主机名作为header，然后使用avro发送给目标机端口。以下是tongjitest165中的flume配置文件：hdfs-agent.sources = avro-collection-sourcehdfs-agent.sinks = hdfs-sink file-sinkhdfs-agent.channels =

47、mem-channel file-channel#hdfs-agent.sources.avro-collection-source.channels = mem-channel file-channelhdfs-agent.sinks.hdfs-sink.channel = mem-channelhdfs-agent.sinks.file-sink.channel = file-channel#指定数据源hdfs-agent.sources.avro-collection-source.type = avrohdfs-agent.sources.avro-collection-source.

48、bind = 65hdfs-agent.sources.avro-collection-source.port = 10000#配置channel hdfs-agent.channels.mem-channel.type= memoryhdfs-agent.channels.mem-channel.capacity = 1000hdfs-agent.channels.mem-channel.transactionCapactiy = 100hdfs-agent.channels.file-channel.type= memoryhdfs-agent.channels.fil

49、e-channel.capacity = 1000hdfs-agent.channels.file-channel.transactionCapactiy = 100#目标是hdfshdfs-agent.sinks.hdfs-sink.type=hdfshdfs-agent.sinks.hdfs-sink.hdfs.path= hdfs:/65:8020/flume/webdatahdfs-agent.sinks.hdfs-sink.hdfs.fileType=DataStreamhdfs-agent.sinks.hdfs-sink.hdfs.batchSize=20hdf

50、s-agent.sinks.hdfs-sink.hdfs.filePrefix=temphdfs-agent.sinks.hdfs-sink.hdfs.rollCount=5#目标是写入本地文件hdfs-agent.sinks.file-sink.type = file_rollhdfs-agent.sinks.file-sink.sink.directory = /home/hadoop/flumehdfs-agent.sinks.file-sink.sink.rollInterval = 0#复用模式hdfs-agent.sources.avro-collection-source.sel

51、ector.type = multiplexinghdfs-agent.sources.avro-collection-source.selector.header=hostnamehdfs-agent.sources.avro-collection-source.selector.mapping.tongjitest166 = file-channelhdfs-agent.sources.avro-collection-source.selector.default = mem-channel上述代码说明从指定的avro源不断读取数据然后根据header的内容发往不同的sink端，本例是如果

52、是tongjitest166来的数据则存储在本地文件中，其他则存储于hdfs中。拦截器的种类有很多，有支持正则表达式的、时间戳的、host、static等种类，可以根据需要自行选择。Flume Sink ProcessorsSink Processor可以实现容错和负载均衡的目的。有Failover Sink Processor、Load balancing Sink Processor、Custom Sink Processor（目前还不支持）。容错和负载均衡机制是以sinkgroups方式实现，需要有多个sink，在选择机制时在，type中指定就可以：failover、load_balan

53、ce。比如：a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 cessor.type = cessor.priority.k1 = 5cessor.priority.k2 = 10cessor.maxpenalty = 10000以上是容错的一个例子，通过设置优先级，当出现问题时会根据优先级选择sink。在sinks中的只要有一个可用就能保证继续正确的运行下去。Flume自定

54、义业务Flume可以根据实际需要实现自己的业务或者扩展source和sink。做到这些，都必须实现四个方法：stop()、start()、process()、configure()，其中处理过程主要在process方法中。可以对其中的一些source和sink做修改：比如：ExecSource类，发现在273行有读取文件的代码line = reader.readLine()，在此可以对line使用java进行一些操作。然后放入event，转而发送到channel中，等待sink消耗event。也可在sink端找到从channel中读取event的代码channel.take()，通过event