首页 > 代码库 > Flume NG 学习笔记(二)单机与集群Flume 配置

Flume NG 学习笔记(二)单机与集群Flume 配置

下面的内容基本来自官网:http://flume.apache.org/FlumeUserGuide.html

本文使用的是最新版本的apache flume 1.5,安装完Flume然后测试下Flume是否可以用,在Flume目录下用以下语句测试:

bin/flume-ng agent -n$agent_name -c conf -f conf/flume-conf.properties.template

结果如图显示:

技术分享


Ok,我们接下去看下面常用架构、功能配置示例

一、最简单的单一代理Flume 配置

 

下面是配置文件:


[html] view plain copy

  1. #文件名:single_case1.conf.conf  

  2. #配置内容:  

  3. #single_case1.conf.conf: A single-node Flume configuration  

  4. #Name the components on this agent  

  5. a1.sourcesr1  

  6. a1.sinksk1  

  7. a1.channelsc1  

  8.    

  9. #Describe/configure the source  

  10. a1.sources.r1.typenetcat  

  11. a1.sources.r1.bindlocalhost  

  12. a1.sources.r1.port44444  

  13.    

  14. #Describe the sink  

  15. a1.sinks.k1.typelogger  

  16.    

  17. #Use a channel which buffers events in memory  

  18. a1.channels.c1.typememory  

  19. a1.channels.c1.capacity1000  

  20. a1.channels.c1.transactionCapacity100  

  21.    

  22. #Bind the source and sink to the channel  

  23. a1.sources.r1.channelsc1  

  24. a1.sinks.k1.channelc1  




说明下,这里所有的例子都是将配置文件放到 $FLUME_HOME/conf 目录下,后面就不赘述了。

 

#敲命令

flume-ng agent -cconf -f conf/single_case1.conf -n a1 -Dflume.root.logger=INFO,console

 

#参数命令

-c conf 指定配置目录为conf

-f conf/single_case1.conf指定配置文件为conf/single_case1.conf

-n a1 指定agent名字为a1,需要与case1_example.conf中的一致

-Dflume.root.logger=INFO,console指定DEBUF模式在console输出INFO信息

具体参数命令请通过flume-nghelp查看

 

#然后在另一个终端进行测试

telnet 127.0.0.1 44444

技术分享


然后会看在之前启动的终端查看console输出到如下:

技术分享


这里会发现消息hello world! 输出了,而hello world! hello world!hello world!则被拦截了。因为在配置文件中,我们选择的输出方式为:a1.sinks.k1.type= logger

,即console输出,flume-ng针对logger是只显示16个字节的,剩下的都被sink截了。下面是源码
在LoggerSink.Java中:


[java] view plain copy

  1. if(event != null) {  

  2.        if (logger.isInfoEnabled()) {  

  3.          logger.info("Event: " + EventHelper.dumpEvent(event));  

  4.        }  

  5. }  



我们去看EventHelper.java的dumpEvent方法:


[java] view plain copy

  1. privatestatic final int DEFAULT_MAX_BYTES = 16;  

  2. publicstatic String dumpEvent(Event event) {  

  3.    return dumpEvent(event, DEFAULT_MAX_BYTES);  

  4. }  

  5.    

  6. publicstatic String dumpEvent(Event event, int maxBytes) {  

  7.    StringBuilder buffer = new StringBuilder();  

  8.    if (event == null || event.getBody() == null) {  

  9.      buffer.append("null");  

  10.    } else if (event.getBody().length == 0) {  

  11.      // do nothing... in this case, HexDump.dump() will throw anexception  

  12.    } else {  

  13.      byte[] body = event.getBody();  

  14.      byte[] data = Arrays.copyOf(body, Math.min(body.length,maxBytes));  

  15.      ByteArrayOutputStream out = new ByteArrayOutputStream();  

  16.      try {  

  17.        HexDump.dump(data, 0, out, 0);  

  18.        String hexDump = new String(out.toByteArray());  

  19.        // remove offset since it‘s not relevant for such a smalldataset  

  20.        if(hexDump.startsWith(HEXDUMP_OFFSET)) {  

  21.          hexDump =hexDump.substring(HEXDUMP_OFFSET.length());  

  22.        }  

  23.        buffer.append(hexDump);  

  24.      } catch (Exception e) {  

  25.       if(LOGGER.isInfoEnabled()) {  

  26.         LOGGER.info("Exception while dumpingevent", e);  

  27.       }  

  28.        buffer.append("...Exception while dumping:").append(e.getMessage());  

  29.      }  

  30.      String result = buffer.toString();  

  31.      if(result.endsWith(EOL) && buffer.length() >EOL.length()) {  

  32.        buffer.delete(buffer.length() - EOL.length(),buffer.length()).toString();  

  33.      }  

  34.    }  

  35.    return "{ headers:" + event.getHeaders() + " body:"+ buffer + " }";  

  36.  }  



不难看出,在event处理过程中,发生了数据截取操作。

Ok,进入下一个环节。

 

二、“集群”代理Flume 配置

技术分享


这里集群的概念是多台机器的管理,最简单的就是两台机器一台代理主机从数据源获取数据,然后将数据在传送到另一台主机上,进行输出。这样做的意义是,一个业务多数据源的时候,我们可以对每个数据源设置代理,然后将它们汇总到一台代理主机上进行输出。

下面实现最简单的集群配置,即两个代理,一台接受数据源数据的代理将数据推送到汇总的代理,而汇总的代理再将数据输出。因此这两台主机分别是push,pull

根据上图需要用AVRO RPC通信,因此推数据sinks类型与拉数据的sources的类型都是avro 。而拉数据代理的数据源,我们用前文讲的Spool Source 形式来处理,这里我们预先建好目录与文件,test.log

技术分享


下面设置推代理主机的flume配置文件:


[html] view plain copy

  1. #推数据代理的配置文件push.conf  

  2. #Name the components on this agent  

  3. a2.sourcesr1  

  4. a2.sinksk1  

  5. a2.channelsc1  

  6.    

  7. #Describe/configure the source  

  8. a2.sources.r1.typespooldir  

  9. a2.sources.r1.spoolDir= /tmp/logs  

  10. a2.sources.r1.channelsc1  

  11.    

  12. #Use a channel which buffers events in memory  

  13. a2.channels.c1.typememory  

  14. a2.channels.c1.keep-alive10  

  15. a2.channels.c1.capacity100000  

  16. a2.channels.c1.transactionCapacity100000  

  17.    

  18. #Describe/configure the source  

  19. a2.sinks.k1.typeavro  

  20. a2.sinks.k1.channelc1  

  21. a2.sinks.k1.hostnamepull  

  22. a2.sinks.k1.port4444  




下面设置汇总代理主机的flume配置文件:


[html] view plain copy

  1. #汇总数据代理的配置文件pull.conf  

  2. #Name the components on this agent  

  3. a1.sourcesr1  

  4. a1.sinksk1  

  5. a1.channelsc1  

  6.    

  7. #Describe/configure the source  

  8. a1.sources.r1.typeavro  

  9. a1.sources.r1.channelsc1  

  10. a1.sources.r1.bindpull  

  11. a1.sources.r1.port44444  

  12.    

  13. #Describe the sink  

  14. a1.sinks.k1.typelogger  

  15.  a1.sinks.k1.channel = c1  

  16.    

  17. #Use a channel which buffers events in memory  

  18. a1.channels.c1.typememory  

  19. a1.channels.c1.keep-alive10  

  20. a1.channels.c1.capacity100000  

  21. a1.channels.c1.transactionCapacity100000  




虽然Spool Source是非实时的,但由于数据量少,处理还是很快的,因此我们只能先启动pull代理。

#敲命令

flume-ng agent -c conf -f conf/pull.conf -n a1 -Dflume.root.logger=INFO,console

技术分享

上图显示成功。

先后去启动push主机的flume

#敲命令

flume-ng agent -n a2 -c conf -f conf/push.conf -Dflume.root.logger=INFO,console

技术分享



查看pull主机的状态,发现数据已经传过来了。

然后会过去看push主机的文件

技术分享

已经加上后缀名.COMPLETED。这与前文说的是一致的。

 

下面只要将新数据存入到目录/tmp/logs,push主机就会将数据发送到pull主机输出,并修改新数据文件的文件名。


本文出自 “先锋的家园” 博客,请务必保留此出处http://jackwxh.blog.51cto.com/2850597/1906866

Flume NG 学习笔记(二)单机与集群Flume 配置