flume 1.8.0 开发基础

介绍

概述

Apache Flume是一个用于高效地从大量异构数据源收集、聚合、传输到一个集中式数据存储的分布式、高可靠、高可用的系统。

Apache Flume是Apache基金会的顶级项目。现在有两个代码版本线可以获取:0.9.x和1.x。本文档对应的是1.x版本。

数据流模型

Event是流经flume agent的最小数据单元。一个Event(由Event接口实现)从source流向channel,再到sink。Event包含了一个payload(byte array)和可选的header(string attributes)。一个flume agent就是一个jvm下的进程:控制着Events从一个外部的源头到一个外部的目的地。

Source消费着具有特殊格式的Events(这些Event传递到Source通过像Web server这样外在的数据源)。例如AvroSource可以被用于接收Avro的Events,从本客户端或者其他运行中的flume客户端。当一个Source接收到一个Event,它会把它插入到一个或者多个Channel里。Channel会被动地存储这些Event直到它们被一个Sink消费到。Flume中一种Channel是FileChannel,其使用文件系统来作为后端存储。Sink需要负责任地将一个Event从Channel中移除,并将其放入像hdfs一样的外部存储系统(例如HDFSEventSink),或者转发到传输中下一个节点的source中。Source和Sink在agent中异步地交互Channel中的Event。

可靠性

Event是存储在Flume agent的Channel里。Sink的责任就是传输Event到下一个agent或者最终的存储系统(像hdfs)。Sink只有当Event写入下一个agent的Channel 或者 存储到最终的系统时才会从channel里面删掉Event。这就是Flume如何在单跳消息传输中提供端到端的可靠性。Flume提供了一个事务性的方法来修复可靠传输中的Event。Source和Sink包含了Event的存储和重试(通过由channel提供的事务)。

构建Flume

获取源码

通过git

编译/测试 Flume

Flume使用maven来build。你可以通过标准的maven命令行来编译Flume。

  1. 仅编译:mvn clean compile
  2. 编译且运行单元测试:mvn clean test
  3. 运行独立的测试:mvn clean test -Dtest=<Test1>,<Test2>,... -DfailIfNoTests=false
  4. 打包:mvn clean install
  5. 打包(忽略单元测试):mvn clean install -DskipTests

注意:Flume build需要在path中有Google Protocol Buffers编译器。

更新Protocol Buffer版本

File channel依赖Protocol Buffer。当你想更新Protocol Buffer版本时,你需要如下更新使用到Protocol Buffer的data access类:

  1. 本机安装你想要的PB版本
  2. 更新pom.xml中PB的版本
  3. 生成flume中新的PB data access类:cd flume-ng-channels/flume-file-channel; mvn -P compile-proto clean package -DskipTests
  4. 在所有生成文件中加上Apache license(如果缺了的话)
  5. rebuild及测试Flume:cd ../..; mvn clean install

开发自定义部分

client

Client在Event产生时运转,并将他们传递到Flume的agent。Client通常运行在应用消费数据的进程空间中。Flume目前支持Avro, log4j, syslog, 以及 Http POST (with a JSON body)方式从外部数据源传输数据。同时ExecSource支持将本地进程的输出作为Flume的输入。

可能已有的方案是不够的。本案例中你可以使用自定义的方法来向flume发送数据。这里有两种方法来实现。第一:写一个自定义的客户端来和flume已有的source交互,像AvroSource 或者 SyslogTcpSource。此时Client需要将数据转换成这些Source能理解的message。另外一个方案:写一个自定义的Flume Source,通过IPC或者RPC,直接地和已有的client应用通信(需要将client的数据转换成Flume的Event)。注意这些存储在flume agent channel中的事件,必须以Flume Event形式存在。

Client SDK

尽管Flume包含了一系列内置的,用于接收数据的方法(即Source),人们常常想直接地通过flume和自定义的程序进行通信。Flume SDK 就是这样一个lib,它可以通过RPC直接地连接到Flume,并且发送到Flume的数据流。

RPC客户端接口

一个RPC客户端接口的实现,包含了支持Flume的RPC方法。用户的程序可以简单地调用Flume SDK客户端的append(Event)或者appendBatch(List<Event>)接口来发送数据,而不用考虑消息交互的细节。用户可以通过使用诸如SimpleEvent类,或者使用EventBuilder的 静态helper方法withBody(),便捷地实现直接提供事件接口所需的事件ARG。

Transaction(事务)接口

Transaction接口是Flume可靠性的基础。所有主要组件(即source,sink和channel)必须使用Flume Transaction。

Transaction在channel的实现中实现。每个source和sink连接到channel时必须要得到一个channnel的对象。Source使用channnelprocessor来管理transaction。sink明确地通过他们配置的channel来管理transaction。存储一个事件(把他们放入channnel中)或者抽取一个事件(从channnel中取出)在一个激活的transaction中完成。例如:

Channel ch = new MemoryChannel();
Transaction txn = ch.getTransaction();
txn.begin();
try {
  // This try clause includes whatever Channel operations you want to do

  Event eventToStage = EventBuilder.withBody("Hello Flume!",
                       Charset.forName("UTF-8"));
  ch.put(eventToStage);
  // Event takenEvent = ch.take();
  // ...
  txn.commit();
} catch (Throwable t) {
  txn.rollback();

  // Log exception, handle individual exceptions as needed

  // re-throw all Errors
  if (t instanceof Error) {
    throw (Error)t;
  }
} finally {
  txn.close();
}

在这里,我们从channel获取transaction。在begin()返回后,Transaction现在处于活动/打开状态,然后将Event放入Channel中。如果put成功,则提交并关闭Transaction。

Sink

Sink的目的就是从Channel中提取事件并将其转发到传输中的下一个Flume Agent或将它们存储在外部存储库中。根据Flume属性文件中的配置,接收器只与一个通道关联。每个已配置的Sink都有一个SinkRunner实例,当Flume框架调用SinkRunner.start()时,会创建一个新线程来驱动Sink(使用SinkRunner.PollingRunner作为线程的Runnable),该线程管理Sink的生命周期。Sink需要实现start()和stop()方法作为LifecycleAware接口的一部分。

  • Sink.start()方法应初始化Sink并将其置于可将事件转发到其下一个目标的状态。
  • Sink.process()应该执行从Channel提取Event并转发它的核心处理过程。
  • Sink.stop()方法应该进行必要的清理(例如释放资源)。

Sink实现还需要实现Configurable接口来处理自己的配置设置。例如:

public class MySink extends AbstractSink implements Configurable {
  private String myProp;

  @Override
  public void configure(Context context) {
    String myProp = context.getString("myProp", "defaultValue");

    // Process the myProp value (e.g. validation)

    // Store myProp for later retrieval by process() method
    this.myProp = myProp;
  }

  @Override
  public void start() {
    // Initialize the connection to the external repository (e.g. HDFS) that
    // this Sink will forward Events to ..
  }

  @Override
  public void stop () {
    // Disconnect from the external respository and do any
    // additional cleanup (e.g. releasing resources or nulling-out
    // field values) ..
  }

  @Override
  public Status process() throws EventDeliveryException {
    Status status = null;

    // Start transaction
    Channel ch = getChannel();
    Transaction txn = ch.getTransaction();
    txn.begin();
    try {
      // This try clause includes whatever Channel operations you want to do

      Event event = ch.take();

      // Send the Event to the external repository.
      // storeSomeData(e);

      txn.commit();
      status = Status.READY;
    } catch (Throwable t) {
      txn.rollback();

      // Log exception, handle individual exceptions as needed

      status = Status.BACKOFF;

      // re-throw all Errors
      if (t instanceof Error) {
        throw (Error)t;
      }
    }
    return status;
  }
}

Source

Source的目的是从外部客户端接收数据并将其存储到已配置的Channels中。Source可以获取其自己的ChannelProcessor的实例来处理在Channel本地事务中提交的串行事件。在exception的情况下,需要Channels传播异常,则所有Channels将回滚其事务,但先前在其他Channel上处理的事件将保持提交。

与SinkRunner.PollingRunner Runnable类似,有一个PollingRunner Runnable,它在Flume框架调用PollableSourceRunner.start()时创建的线程上执行。每个配置的PollableSource都与自己运行PollingRunner的线程相关联。该线程管理PollableSource的生命周期,例如启动和停止。

  • PollableSource必须实现LifecycleAware接口中声明的start()和stop()方法。
  • PollableSource的运行器调用Source的process()方法。 process()方法应检查新数据并将其作为Flume事件存储到Channel中。

注意,实际上有两种类型的Source:已经提到过PollableSource,另一个是EventDrivenSource。与PollableSource不同,EventDrivenSource必须有自己的回调机制,捕获新数据并将其存储到Channel中。EventDrivenSources并不像PollableSources那样由它们自己的线程驱动。下面是一个自定义PollableSource的示例:

public class MySource extends AbstractSource implements Configurable, PollableSource {
  private String myProp;

  @Override
  public void configure(Context context) {
    String myProp = context.getString("myProp", "defaultValue");

    // Process the myProp value (e.g. validation, convert to another type, ...)

    // Store myProp for later retrieval by process() method
    this.myProp = myProp;
  }

  @Override
  public void start() {
    // Initialize the connection to the external client
  }

  @Override
  public void stop () {
    // Disconnect from external client and do any additional cleanup
    // (e.g. releasing resources or nulling-out field values) ..
  }

  @Override
  public Status process() throws EventDeliveryException {
    Status status = null;

    try {
      // This try clause includes whatever Channel/Event operations you want to do

      // Receive new data
      Event e = getSomeData();

      // Store the Event into this Source's associated Channel(s)
      getChannelProcessor().processEvent(e);

      status = Status.READY;
    } catch (Throwable t) {
      // Log exception, handle individual exceptions as needed

      status = Status.BACKOFF;

      // re-throw all Errors
      if (t instanceof Error) {
        throw (Error)t;
      }
    } finally {
      txn.close();
    }
    return status;
  }
}

参考自(Flume 1.8.0 Developer Guide)

flume 1.8.0 文档完整翻译可见 https://blog.csdn.net/u013128262

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏吴伟祥

使用Idea迅速构建一个Spring Boot应用与部署

1、选择File->New->Project,在对话框中选中Spring Initializr,给项目配置合适的SDK,点击Next.

2274
来自专栏Gaussic

基于Spring Boot 的Blog开发 原

之前写过几篇关于利用Spring MVC来开发的博客,从博客下面的评论以及GitHub上的Issues看还是会出现许多的问题,且大部分的问题都出在配置上。虽然说...

991
来自专栏Danny的专栏

【EJB学习笔记】——EJB开发环境搭建(Eclipse集成JBoss)

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/huyuyang6688/article/...

1473
来自专栏玩转JavaEE

初识Spring Boot框架

按:最近公众号文章主要是整理一些老文章,主要是个人CSDN上的博客,也会穿插一些新的技术点。 ---- 前面的铺垫文章已经连着写了六篇了,主要是介绍了Sprin...

3845
来自专栏Java后端技术

context:component-scan标签的use-default-filters属性的作用以及原理分析

  我们在Spring+SpringMVC+Mybatis的集成开发中,经常会遇到事务配置不起作用等问题,那么本文就来分析下出现这种问题可能的原因以及解决方式。

904
来自专栏一个会写诗的程序员的博客

《Spring Boot 实战开发》 附录 II : Spring Boot 2.0 新特性《Spring Boot 实战开发》 附录 II : Spring Boot 2.0 新特性Kotlin

本节我们介绍 Spring Boot 2.0 版本的众多新特性,内容包括了 M1~M7里程碑版本的核心新功能特性。不过,我们首先把对 Kotlin 的特性的支持...

1693
来自专栏清晨我上码

第四节 微服务OTRS SpringCould使用

1.1. 启动一个服务注册中心,只需要一个注解@EnableEurekaServer,这个注解需要在springboot工程的启动application类上...

1033
来自专栏微信公众号:Java团长

初识Spring Boot框架

前面的铺垫文章已经写了几篇了,主要是介绍了Spring和SpringMVC框架,小伙伴们在学习的过程中大概也发现了这两个框架需要我们手动配置的地方非常多,不过做...

1104
来自专栏ImportSource

Spring Boot 2.0 新特性和发展方向

以Java 8 为基准 Spring Boot 2.0 要求Java 版本必须8以上, Java 6 和 7 不再支持。 内嵌容器包结构调整 为了支持react...

4439
来自专栏Android 研究

Android JNI学习(二)——实战JNI之“hello world”

首先在Android Studio创建一个Android项目,包名为gebilaolitou.ndkdemo

2834

扫码关注云+社区

领取腾讯云代金券