OneCoder Avatar
OneCodercoderli.com · 938 篇博文
🐍 Python

Python 与大数据

MySQL、Hadoop、自动化与基础工具

🎨 视觉封面

Flume(ng) 自定义sink实现和属性注入

📅 2013-10-29·✍️ onecoder·计算中...·⏱️ 4 分钟
#Flume

最近需要利用flume来做收集远端日志,所以学习一些flume最基本的用法。这里仅作记录。

远端日志收集的整体思路是远端自定义实现log4j的appender把消息发送到flume端,flume端自定义实现一个sink来按照我们的规则保存日志。

自定义Sink代码:

JAVA
public class LocalFileLogSink extends AbstractSink implements Configurable {
     private static final Logger logger = LoggerFactory
              . getLogger(LocalFileLogSink .class );
            private static final String PROP_KEY_ROOTPATH = "rootPath";
      private String rootPath;
     @Override
     public void configure(Context context) {
          String rootPath = context.getString(PROP_KEY_ROOTPATH );
          setRootPath(rootPath);
     }
          
          @Override
          public Status process() throws EventDeliveryException {
           logger .debug("Do process" );

}

实现Configurable接口,即可在初始化时,通过configure方法从context中获取配置的参数的值。这里,我们是想从flume的配置文件中获取rootPath的值,也就是日志保存的根路径。在flume-conf.properties中配置如下:

TEXT
agent.sinks = loggerSink
agent.sinks.loggerSink.rootPath = ./logs

loggerSink是自定义sink的名称,我们取值时的key,只需要loggerSink后面的部分即可,即这里的rootPath。

实际业务逻辑的执行,是通过继承复写AbstractSink中的process方法实现。从基类的getChannel方法中获取信道,从中取出Event处理即可。

JAVA
 Channel ch = getChannel();
            Transaction txn = ch.getTransaction();
          txn.begin();
           try {
               logger .debug("Get event." );
              Event event = ch.take();
              txn.commit();
              status = Status. READY ;
              return status;
finally {
              Log. info( "trx close.");
              txn.close();
          }
💡 OneCoder 资源指引

所有代码开源上传至 GitHub:yummy-code 仓库 · GESP 专题站:GESP WIKI

🤝 技术交流与答疑

欢迎加入:C++ GESP/CSP 考级答疑群(688906745)Java/Python交流群(982860385),点击可直接加群。

📚

猜你想读 · 相关文章推荐

OneCoder

OneCoder (lihongzheshuai)

一个中年人的自留地,记录学习 C++、GESP/NOI、Java、Python 与算法架构的心得体会。本站唯一网址:coderli.com

💬 读者留言与交流

0 条讨论
✨ 支持 Markdown 语法格式
还没有留言,快来成为第一个讨论者吧!