本文由云+社区发表作者：皮皮熊概述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编译/测试 FlumeFlume使用maven来build。你可以通过标准的maven命令行来编译Flume。仅编译:mvn clean compile编译且运行单元测试:mvn clean test运行独立的测试:mvn clean test -Dtest=<Test1>,<Test2>,… -DfailIfNoTests=false打包:mvn clean install打包(忽略单元测试):mvn clean install -DskipTests注意：Flume build需要在path中有Google Protocol Buffers编译器。更新Protocol Buffer版本File channel依赖Protocol Buffer。当你想更新Protocol Buffer版本时，你需要如下更新使用到Protocol Buffer的data access类：本机安装你想要的PB版本更新pom.xml中PB的版本生成flume中新的PB data access类：cd flume-ng-channels/flume-file-channel; mvn -P compile-proto clean package -DskipTests在所有生成文件中加上Apache license(如果缺了的话)rebuild及测试Flume:cd ../..; mvn clean install开发自定义部分clientClient在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。SinkSink的目的就是从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; }}SourceSource的目的是从外部客户端接收数据并将其存储到已配置的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此文已由腾讯云+社区在各渠道发布获取更多新鲜技术干货，可以关注我们腾讯云技术社区-云加社区官方号及知乎机构号
...