zoukankan      html  css  js  c++  java
  • flume ng之TailSource

    在它里面自带了一个TailSource以及TailDirSource,这个Source是负责读取一个文件,并一行一行的发送到sink端,而在flume-ng 1.4.0里面没有自带TailSource,更别说TailDirSource了,虽然我们可以在flume-ng 1.4.0里面用exec的tail -F来一行一行的发送文件,但是还是不太好。而且公司最近flume需要升级,之前的程序用到了TailSource和TailDirSource,没办法只能再次开发了。在flume-ng里面开发一个source需要遵循一定的规则,我们可以通过以下两种方式来开发自己的source:

    分布式:hadoop、hive、spark、shark、flume技术

    public class TailSource extends AbstractSource implments EventDrivenSource, configurable {

      @Override

      public void configure(Context context) {

      }

      @Override

      public synchronized void start() {

        

      }

      @Override

      public synchronized void stop() {

      }

    }

    或者

    public class TailSource extends AbstractSource implements Configurable, PollableSource {

      @Override

      public void configure(Context context) {

      }

      @Override

      public Status process() throws EventDelieryException {

        return null;

      }

      @Override

      public synchronized void start() {

      }

      @Override

      public synchronized void stop() {

      }

    }

        这两个类虽然继承关系不一样,但是都可以实现。下面的方法中多了一个process()方法,这个方法在PollableSourceRunner类中会被一个线程不断的调用。

        关于如何具体开发TailSource,这个我就不介绍了。下面主要介绍一下如何使用我这个TailSource,在本文的后面会附上TailSource插件的下载地址,大家如有需要可以自行下载使用。TailSource支持waitTime、file、startFromEnd以及offset四个参数的配置。

        waitTime:指的是隔多久读一次文件,程序是按照一块一块的读取文件内容到缓冲区,然后一行一行的解析;可以不配置,默认一秒读一次。

        file:指的是需要读取的文件绝对路径,一定需要配置;

        startFromEnd:是否从文件结尾处读取,默认为false;

        offset:从offset偏移量开始读取,可以不配置,默认是从偏移量为0开始读。

    目前版本的TailSource不支持正则表达式的配置。如你需要使用,可以按照下面的配置进行:

    agent.sources = r1

    agent.sources.r1.type = org.apache.flume.TailSource

    agent.sources.r1.channels = channel

    agent.sources.r1.waitTime = 100

    agent.sources.r1.file = /export1/spark/apache-flume-1.4.0-bin/test

    agent.sources.r1.startFromEnd = false

    agent.sources.r1.offset = 0

    然后就可以监听到test文件的变化,并放到channel里面。

    日志依赖的jar包在maven中的配置如下:

    <dependency>

      <groupId>org.slf4j</groupId>

      <artifactId>slf4j-api</artifactId>

    </dependency>

    <dependency>

      <groupId>org.slf4j</groupId>

      <artifactId>slf4j-log4j12</artifactId>

      <scope>test</scope>

    </dependency>

    flume里jar包在maven中的配置如下:

    <dependency>

      <groupId>org.slf4j</groupId>

      <artifactId>slf4j-api</artifactId>

    </dependency>

    <dependency>

      <groupId>org.apache.flume</groupId>

      <artifactId>flume-ng-sdk</artifactId>

    </dependency>

    <dependency>

      <groupId>org.apache.flume</groupId>

      <artifactId>flume-ng-core</artifactId>

    </dependency>

    <dependency>

      <groupId>org.apache.flume</groupId>

      <artifactId>flume-ng-configuration</artifactId>

    </dependency>

  • 相关阅读:
    CDN实现原理
    openstack测试集群扩容配置
    ELK5.2.2自动化部署脚本
    Python多线程
    Python反射
    OpenStack快照分析:(三)从磁盘启动云主机离线(在线)快照分析
    OpenStack快照分析:(二)从镜像启动的云主机离在线快照分析
    OpenStack快照分析:(一)从镜像启动的云主机离线快照分析
    Python元编程
    Python源码阅读:对象
  • 原文地址:https://www.cnblogs.com/man-li/p/4362005.html
Copyright © 2011-2022 走看看