content/FlumeDeveloperGuide.html

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> <html xmlns="http://www.w3.org/1999/xhtml"> <head> <meta http-equiv="Content-Type" content="text/html; charset=utf-8" /> <title>Flume 1.11.0 Developer Guide — Apache Flume</title> <link rel="stylesheet" href="_static/flume.css" type="text/css" /> <link rel="stylesheet" href="_static/pygments.css" type="text/css" /> <script type="text/javascript"> var DOCUMENTATION_OPTIONS = { URL_ROOT: '', VERSION: '', COLLAPSE_INDEX: false, FILE_SUFFIX: '.html', HAS_SOURCE: true }; </script> <script type="text/javascript" src="_static/jquery.js"></script> <script type="text/javascript" src="_static/underscore.js"></script> <script type="text/javascript" src="_static/doctools.js"></script> <link rel="top" title="Apache Flume" href="index.html" /> <link rel="up" title="Documentation" href="documentation.html" /> <link rel="next" title="Releases" href="releases/index.html" /> <link rel="prev" title="Flume 1.11.0 User Guide" href="FlumeUserGuide.html" /> </head> <body> <div class="header"> <table width="100%" border="0"> <tr> <td width="10%"> <div class="logo"> <a href="index.html"> <img class="logo" src="_static/flume-logo.png" alt="Logo"/> </a> </div> </td> <td width="2%"> ™ </td> <td width="68%" align="center" class="pageTitle">Apache Flume™ </td> <td width="20%"> <a href="https://www.apache.org"> <img src="_static/feather-small.png" alt="Apache Software Foundation" height="70"/> </a> </td> </tr> </table> </div> <div class="document"> <div class="documentwrapper"> <div class="bodywrapper"> <div class="body"> <div class="section" id="flume-1-11-0-developer-guide"> <h1>Flume 1.11.0 Developer Guide<a class="headerlink" href="#flume-1-11-0-developer-guide" title="Permalink to this headline">¶</a></h1> <div class="section" id="introduction"> <h2>Introduction<a class="headerlink" href="#introduction" title="Permalink to this headline">¶</a></h2> <div class="section" id="overview"> <h3>Overview<a class="headerlink" href="#overview" title="Permalink to this headline">¶</a></h3> Apache Flume is a distributed, reliable, and available system for efficiently collecting, aggregating and moving large amounts of log data from many different sources to a centralized data store. Apache Flume is a top-level project at the Apache Software Foundation. There are currently two release code lines available, versions 0.9.x and 1.x. This documentation applies to the 1.x codeline. For the 0.9.x codeline, please see the <a class="reference external" href="https://archive.cloudera.com/cdh/3/flume/DeveloperGuide/">Flume 0.9.x Developer Guide</a>. </div> <div class="section" id="architecture"> <h3>Architecture<a class="headerlink" href="#architecture" title="Permalink to this headline">¶</a></h3> <div class="section" id="data-flow-model"> <h4>Data flow model<a class="headerlink" href="#data-flow-model" title="Permalink to this headline">¶</a></h4> An <tt class="docutils literal">Event</tt> is a unit of data that flows through a Flume agent. The <tt class="docutils literal">Event</tt> flows from <tt class="docutils literal">Source</tt> to <tt class="docutils literal">Channel</tt> to <tt class="docutils literal">Sink</tt>, and is represented by an implementation of the <tt class="docutils literal">Event</tt> interface. An <tt class="docutils literal">Event</tt> carries a payload (byte array) that is accompanied by an optional set of headers (string attributes). A Flume agent is a process (JVM) that hosts the components that allow <tt class="docutils literal">Event</tt>s to flow from an external source to a external destination. <div class="figure align-center"> <img alt="Agent component diagram" src="_images/DevGuide_image00.png" /> </div> A <tt class="docutils literal">Source</tt> consumes <tt class="docutils literal">Event</tt>s having a specific format, and those <tt class="docutils literal">Event</tt>s are delivered to the <tt class="docutils literal">Source</tt> by an external source like a web server. For example, an <tt class="docutils literal">AvroSource</tt> can be used to receive Avro <tt class="docutils literal">Event</tt>s from clients or from other Flume agents in the flow. When a <tt class="docutils literal">Source</tt> receives an <tt class="docutils literal">Event</tt>, it stores it into one or more <tt class="docutils literal">Channel</tt>s. The <tt class="docutils literal">Channel</tt> is a passive store that holds the <tt class="docutils literal">Event</tt> until that <tt class="docutils literal">Event</tt> is consumed by a <tt class="docutils literal">Sink</tt>. One type of <tt class="docutils literal">Channel</tt> available in Flume is the <tt class="docutils literal">FileChannel</tt> which uses the local filesystem as its backing store. A <tt class="docutils literal">Sink</tt> is responsible for removing an <tt class="docutils literal">Event</tt> from the <tt class="docutils literal">Channel</tt> and putting it into an external repository like HDFS (in the case of an <tt class="docutils literal">HDFSEventSink</tt>) or forwarding it to the <tt class="docutils literal">Source</tt> at the next hop of the flow. The <tt class="docutils literal">Source</tt> and <tt class="docutils literal">Sink</tt> within the given agent run asynchronously with the <tt class="docutils literal">Event</tt>s staged in the <tt class="docutils literal">Channel</tt>. </div> <div class="section" id="reliability"> <h4>Reliability<a class="headerlink" href="#reliability" title="Permalink to this headline">¶</a></h4> An <tt class="docutils literal">Event</tt> is staged in a Flume agent’s <tt class="docutils literal">Channel</tt>. Then it’s the <tt class="docutils literal">Sink</tt>‘s responsibility to deliver the <tt class="docutils literal">Event</tt> to the next agent or terminal repository (like HDFS) in the flow. The <tt class="docutils literal">Sink</tt> removes an <tt class="docutils literal">Event</tt> from the <tt class="docutils literal">Channel</tt> only after the <tt class="docutils literal">Event</tt> is stored into the <tt class="docutils literal">Channel</tt> of the next agent or stored in the terminal repository. This is how the single-hop message delivery semantics in Flume provide end-to-end reliability of the flow. Flume uses a transactional approach to guarantee the reliable delivery of the <tt class="docutils literal">Event</tt>s. The <tt class="docutils literal">Source</tt>s and <tt class="docutils literal">Sink</tt>s encapsulate the storage/retrieval of the <tt class="docutils literal">Event</tt>s in a <tt class="docutils literal">Transaction</tt> provided by the <tt class="docutils literal">Channel</tt>. This ensures that the set of <tt class="docutils literal">Event</tt>s are reliably passed from point to point in the flow. In the case of a multi-hop flow, the <tt class="docutils literal">Sink</tt> from the previous hop and the <tt class="docutils literal">Source</tt> of the next hop both have their <tt class="docutils literal">Transaction</tt>s open to ensure that the <tt class="docutils literal">Event</tt> data is safely stored in the <tt class="docutils literal">Channel</tt> of the next hop. </div> </div> <div class="section" id="building-flume"> <h3>Building Flume<a class="headerlink" href="#building-flume" title="Permalink to this headline">¶</a></h3> <div class="section" id="getting-the-source"> <h4>Getting the source<a class="headerlink" href="#getting-the-source" title="Permalink to this headline">¶</a></h4> Check-out the code using Git. Click here for <a class="reference external" href="https://git-wip-us.apache.org/repos/asf/flume.git">the git repository root</a> or at <a class="reference external" href="https://github.com/apache/flume.git">GitHub</a>. The Flume 1.x development happens under the branch “trunk” so this command line can be used: <blockquote> <div>git clone <a class="reference external" href="https://git-wip-us.apache.org/repos/asf/flume.git">https://git-wip-us.apache.org/repos/asf/flume.git</a></div></blockquote> or <blockquote> <div>git clone <a class="reference external" href="https://github.com/apache/flume.git">https://github.com/apache/flume.git</a></div></blockquote> </div> <div class="section" id="compile-test-flume"> <h4>Compile/test Flume<a class="headerlink" href="#compile-test-flume" title="Permalink to this headline">¶</a></h4> The Flume build is mavenized. You can compile Flume using the standard Maven commands: <ol class="arabic simple"> <li>Compile only: <tt class="docutils literal">mvn clean compile</tt></li> <li>Compile and run unit tests: <tt class="docutils literal">mvn clean test</tt></li> <li>Run individual test(s): <tt class="docutils literal">mvn clean test -Dtest=<Test1>,<Test2>,... -DfailIfNoTests=false</tt></li> <li>Create tarball package: <tt class="docutils literal">mvn clean install</tt></li> <li>Create tarball package (skip unit tests): <tt class="docutils literal">mvn clean install -DskipTests</tt></li> </ol> </div> <div class="section" id="updating-protocol-buffer-version"> <h4>Updating Protocol Buffer Version<a class="headerlink" href="#updating-protocol-buffer-version" title="Permalink to this headline">¶</a></h4> File channel has a dependency on Protocol Buffer. When updating the version of Protocol Buffer used by Flume, it is necessary to regenerate the data access classes using the protoc compiler that is part of Protocol Buffer as follows. <ol class="arabic simple"> <li>Update version of Protocol Buffer in pom.xml</li> <li>Add Apache license header to any of the generated files that are missing it</li> <li>Rebuild and test Flume: <tt class="docutils literal">cd ../..; mvn clean install</tt></li> </ol> </div> </div> <div class="section" id="developing-custom-components"> <h3>Developing custom components<a class="headerlink" href="#developing-custom-components" title="Permalink to this headline">¶</a></h3> <div class="section" id="client"> <h4>Client<a class="headerlink" href="#client" title="Permalink to this headline">¶</a></h4> The client operates at the point of origin of events and delivers them to a Flume agent. Clients typically operate in the process space of the application they are consuming data from. Flume currently supports Avro, log4j, syslog, and Http POST (with a JSON body) as ways to transfer data from a external source. Additionally, there’s an <tt class="docutils literal">ExecSource</tt> that can consume the output of a local process as input to Flume. It’s quite possible to have a use case where these existing options are not sufficient. In this case you can build a custom mechanism to send data to Flume. There are two ways of achieving this. The first option is to create a custom client that communicates with one of Flume’s existing <tt class="docutils literal">Source</tt>s like <tt class="docutils literal">AvroSource</tt> or <tt class="docutils literal">SyslogTcpSource</tt>. Here the client should convert its data into messages understood by these Flume <tt class="docutils literal">Source</tt>s. The other option is to write a custom Flume <tt class="docutils literal">Source</tt> that directly talks with your existing client application using some IPC or RPC protocol, and then converts the client data into Flume <tt class="docutils literal">Event</tt>s to be sent downstream. Note that all events stored within the <tt class="docutils literal">Channel</tt> of a Flume agent must exist as Flume <tt class="docutils literal">Event</tt>s. <div class="section" id="client-sdk"> <h5>Client SDK<a class="headerlink" href="#client-sdk" title="Permalink to this headline">¶</a></h5> Though Flume contains a number of built-in mechanisms (i.e. <tt class="docutils literal">Source</tt>s) to ingest data, often one wants the ability to communicate with Flume directly from a custom application. The Flume Client SDK is a library that enables applications to connect to Flume and send data into Flume’s data flow over RPC. </div> <div class="section" id="rpc-client-interface"> <h5>RPC client interface<a class="headerlink" href="#rpc-client-interface" title="Permalink to this headline">¶</a></h5> An implementation of Flume’s RpcClient interface encapsulates the RPC mechanism supported by Flume. The user’s application can simply call the Flume Client SDK’s <tt class="docutils literal">append(Event)</tt> or <tt class="docutils literal">appendBatch(List<Event>)</tt> to send data and not worry about the underlying message exchange details. The user can provide the required <tt class="docutils literal">Event</tt> arg by either directly implementing the <tt class="docutils literal">Event</tt> interface, by using a convenience implementation such as the SimpleEvent class, or by using <tt class="docutils literal">EventBuilder</tt>‘s overloaded <tt class="docutils literal">withBody()</tt> static helper methods. </div> <div class="section" id="rpc-clients-avro-and-thrift"> <h5>RPC clients - Avro and Thrift<a class="headerlink" href="#rpc-clients-avro-and-thrift" title="Permalink to this headline">¶</a></h5> As of Flume 1.4.0, Avro is the default RPC protocol. The <tt class="docutils literal">NettyAvroRpcClient</tt> and <tt class="docutils literal">ThriftRpcClient</tt> implement the <tt class="docutils literal">RpcClient</tt> interface. The client needs to create this object with the host and port of the target Flume agent, and can then use the <tt class="docutils literal">RpcClient</tt> to send data into the agent. The following example shows how to use the Flume Client SDK API within a user’s data-generating application: <div class="highlight-java"><div class="highlight"><pre>import org.apache.flume.Event; import org.apache.flume.EventDeliveryException; import org.apache.flume.api.RpcClient; import org.apache.flume.api.RpcClientFactory; import org.apache.flume.event.EventBuilder; import java.nio.charset.Charset; public class MyApp { public static void main(String[] args) { MyRpcClientFacade client = new MyRpcClientFacade(); // Initialize client with the remote Flume agent's host and port client.init("host.example.org", 41414); // Send 10 events to the remote Flume agent. That agent should be // configured to listen with an AvroSource. String sampleData = "Hello Flume!"; for (int i = 0; i < 10; i++) { client.sendDataToFlume(sampleData); } client.cleanUp(); } } class MyRpcClientFacade { private RpcClient client; private String hostname; private int port; public void init(String hostname, int port) { // Setup the RPC connection this.hostname = hostname; this.port = port; this.client = RpcClientFactory.getDefaultInstance(hostname, port); // Use the following method to create a thrift client (instead of the above line): // this.client = RpcClientFactory.getThriftInstance(hostname, port); } public void sendDataToFlume(String data) { // Create a Flume Event object that encapsulates the sample data Event event = EventBuilder.withBody(data, Charset.forName("UTF-8")); // Send the event try { client.append(event); } catch (EventDeliveryException e) { // clean up and recreate the client client.close(); client = null; client = RpcClientFactory.getDefaultInstance(hostname, port); // Use the following method to create a thrift client (instead of the above line): // this.client = RpcClientFactory.getThriftInstance(hostname, port); } } public void cleanUp() { // Close the RPC connection client.close(); } } </pre></div> </div> The remote Flume agent needs to have an <tt class="docutils literal">AvroSource</tt> (or a <tt class="docutils literal">ThriftSource</tt> if you are using a Thrift client) listening on some port. Below is an example Flume agent configuration that’s waiting for a connection from MyApp: <div class="highlight-properties"><div class="highlight"><pre>a1.channels = c1 a1.sources = r1 a1.sinks = k1 a1.channels.c1.type = memory a1.sources.r1.channels = c1 a1.sources.r1.type = avro # For using a thrift source set the following instead of the above line. # a1.source.r1.type = thrift a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 41414 a1.sinks.k1.channel = c1 a1.sinks.k1.type = logger </pre></div> </div> For more flexibility, the default Flume client implementations (<tt class="docutils literal">NettyAvroRpcClient</tt> and <tt class="docutils literal">ThriftRpcClient</tt>) can be configured with these properties: <div class="highlight-properties"><div class="highlight"><pre>client.type = default (for avro) or thrift (for thrift) hosts = h1 # default client accepts only 1 host # (additional hosts will be ignored) hosts.h1 = host1.example.org:41414 # host and port must both be specified # (neither has a default) batch-size = 100 # Must be >=1 (default: 100) connect-timeout = 20000 # Must be >=1000 (default: 20000) request-timeout = 20000 # Must be >=1000 (default: 20000) </pre></div> </div> </div> <div class="section" id="secure-rpc-client-thrift"> <h5>Secure RPC client - Thrift<a class="headerlink" href="#secure-rpc-client-thrift" title="Permalink to this headline">¶</a></h5> As of Flume 1.6.0, Thrift source and sink supports kerberos based authentication. The client needs to use the getThriftInstance method of <tt class="docutils literal">SecureRpcClientFactory</tt> to get hold of a <tt class="docutils literal">SecureThriftRpcClient</tt>. <tt class="docutils literal">SecureThriftRpcClient</tt> extends <tt class="docutils literal">ThriftRpcClient</tt> which implements the <tt class="docutils literal">RpcClient</tt> interface. The kerberos authentication module resides in flume-ng-auth module which is required in classpath, when using the <tt class="docutils literal">SecureRpcClientFactory</tt>. Both the client principal and the client keytab should be passed in as parameters through the properties and they reflect the credentials of the client to authenticate against the kerberos KDC. In addition, the server principal of the destination Thrift source to which this client is connecting to, should also be provided. The following example shows how to use the <tt class="docutils literal">SecureRpcClientFactory</tt> within a user’s data-generating application: <div class="highlight-java"><div class="highlight"><pre>import org.apache.flume.Event; import org.apache.flume.EventDeliveryException; import org.apache.flume.event.EventBuilder; import org.apache.flume.api.SecureRpcClientFactory; import org.apache.flume.api.RpcClientConfigurationConstants; import org.apache.flume.api.RpcClient; import java.nio.charset.Charset; import java.util.Properties; public class MyApp { public static void main(String[] args) { MySecureRpcClientFacade client = new MySecureRpcClientFacade(); // Initialize client with the remote Flume agent's host, port Properties props = new Properties(); props.setProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, "thrift"); props.setProperty("hosts", "h1"); props.setProperty("hosts.h1", "client.example.org"+":"+ String.valueOf(41414)); // Initialize client with the kerberos authentication related properties props.setProperty("kerberos", "true"); props.setProperty("client-principal", "flumeclient/client.example.org@EXAMPLE.ORG"); props.setProperty("client-keytab", "/tmp/flumeclient.keytab"); props.setProperty("server-principal", "flume/server.example.org@EXAMPLE.ORG"); client.init(props); // Send 10 events to the remote Flume agent. That agent should be // configured to listen with an AvroSource. String sampleData = "Hello Flume!"; for (int i = 0; i < 10; i++) { client.sendDataToFlume(sampleData); } client.cleanUp(); } } class MySecureRpcClientFacade { private RpcClient client; private Properties properties; public void init(Properties properties) { // Setup the RPC connection this.properties = properties; // Create the ThriftSecureRpcClient instance by using SecureRpcClientFactory this.client = SecureRpcClientFactory.getThriftInstance(properties); } public void sendDataToFlume(String data) { // Create a Flume Event object that encapsulates the sample data Event event = EventBuilder.withBody(data, Charset.forName("UTF-8")); // Send the event try { client.append(event); } catch (EventDeliveryException e) { // clean up and recreate the client client.close(); client = null; client = SecureRpcClientFactory.getThriftInstance(properties); } } public void cleanUp() { // Close the RPC connection client.close(); } } </pre></div> </div> The remote <tt class="docutils literal">ThriftSource</tt> should be started in kerberos mode. Below is an example Flume agent configuration that’s waiting for a connection from MyApp: <div class="highlight-properties"><div class="highlight"><pre>a1.channels = c1 a1.sources = r1 a1.sinks = k1 a1.channels.c1.type = memory a1.sources.r1.channels = c1 a1.sources.r1.type = thrift a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 41414 a1.sources.r1.kerberos = true a1.sources.r1.agent-principal = flume/server.example.org@EXAMPLE.ORG a1.sources.r1.agent-keytab = /tmp/flume.keytab a1.sinks.k1.channel = c1 a1.sinks.k1.type = logger </pre></div> </div> </div> <div class="section" id="failover-client"> <h5>Failover Client<a class="headerlink" href="#failover-client" title="Permalink to this headline">¶</a></h5> This class wraps the default Avro RPC client to provide failover handling capability to clients. This takes a whitespace-separated list of <host>:<port> representing the Flume agents that make-up a failover group. The Failover RPC Client currently does not support thrift. If there’s a communication error with the currently selected host (i.e. agent) agent, then the failover client automatically fails-over to the next host in the list. For example: <div class="highlight-java"><div class="highlight"><pre>// Setup properties for the failover Properties props = new Properties(); props.put("client.type", "default_failover"); // List of hosts (space-separated list of user-chosen host aliases) props.put("hosts", "h1 h2 h3"); // host/port pair for each host alias String host1 = "host1.example.org:41414"; String host2 = "host2.example.org:41414"; String host3 = "host3.example.org:41414"; props.put("hosts.h1", host1); props.put("hosts.h2", host2); props.put("hosts.h3", host3); // create the client with failover properties RpcClient client = RpcClientFactory.getInstance(props); </pre></div> </div> For more flexibility, the failover Flume client implementation (<tt class="docutils literal">FailoverRpcClient</tt>) can be configured with these properties: <div class="highlight-properties"><div class="highlight"><pre>client.type = default_failover hosts = h1 h2 h3 # at least one is required, but 2 or # more makes better sense hosts.h1 = host1.example.org:41414 hosts.h2 = host2.example.org:41414 hosts.h3 = host3.example.org:41414 max-attempts = 3 # Must be >=0 (default: number of hosts # specified, 3 in this case). A '0' # value doesn't make much sense because # it will just cause an append call to # immmediately fail. A '1' value means # that the failover client will try only # once to send the Event, and if it # fails then there will be no failover # to a second client, so this value # causes the failover client to # degenerate into just a default client. # It makes sense to set this value to at # least the number of hosts that you # specified. batch-size = 100 # Must be >=1 (default: 100) connect-timeout = 20000 # Must be >=1000 (default: 20000) request-timeout = 20000 # Must be >=1000 (default: 20000) </pre></div> </div> </div> <div class="section" id="loadbalancing-rpc-client"> <h5>LoadBalancing RPC client<a class="headerlink" href="#loadbalancing-rpc-client" title="Permalink to this headline">¶</a></h5> The Flume Client SDK also supports an RpcClient which load-balances among multiple hosts. This type of client takes a whitespace-separated list of <host>:<port> representing the Flume agents that make-up a load-balancing group. This client can be configured with a load balancing strategy that either randomly selects one of the configured hosts, or selects a host in a round-robin fashion. You can also specify your own custom class that implements the <tt class="docutils literal">LoadBalancingRpcClient$HostSelector</tt> interface so that a custom selection order is used. In that case, the FQCN of the custom class needs to be specified as the value of the <tt class="docutils literal">host-selector</tt> property. The LoadBalancing RPC Client currently does not support thrift. If <tt class="docutils literal">backoff</tt> is enabled then the client will temporarily blacklist hosts that fail, causing them to be excluded from being selected as a failover host until a given timeout. When the timeout elapses, if the host is still unresponsive then this is considered a sequential failure, and the timeout is increased exponentially to avoid potentially getting stuck in long waits on unresponsive hosts. The maximum backoff time can be configured by setting <tt class="docutils literal">maxBackoff</tt> (in milliseconds). The maxBackoff default is 30 seconds (specified in the <tt class="docutils literal">OrderSelector</tt> class that’s the superclass of both load balancing strategies). The backoff timeout will increase exponentially with each sequential failure up to the maximum possible backoff timeout. The maximum possible backoff is limited to 65536 seconds (about 18.2 hours). For example: <div class="highlight-java"><div class="highlight"><pre>// Setup properties for the load balancing Properties props = new Properties(); props.put("client.type", "default_loadbalance"); // List of hosts (space-separated list of user-chosen host aliases) props.put("hosts", "h1 h2 h3"); // host/port pair for each host alias String host1 = "host1.example.org:41414"; String host2 = "host2.example.org:41414"; String host3 = "host3.example.org:41414"; props.put("hosts.h1", host1); props.put("hosts.h2", host2); props.put("hosts.h3", host3); props.put("host-selector", "random"); // For random host selection // props.put("host-selector", "round_robin"); // For round-robin host // // selection props.put("backoff", "true"); // Disabled by default. props.put("maxBackoff", "10000"); // Defaults 0, which effectively // becomes 30000 ms // Create the client with load balancing properties RpcClient client = RpcClientFactory.getInstance(props); </pre></div> </div> For more flexibility, the load-balancing Flume client implementation (<tt class="docutils literal">LoadBalancingRpcClient</tt>) can be configured with these properties: <div class="highlight-properties"><div class="highlight"><pre>client.type = default_loadbalance hosts = h1 h2 h3 # At least 2 hosts are required hosts.h1 = host1.example.org:41414 hosts.h2 = host2.example.org:41414 hosts.h3 = host3.example.org:41414 backoff = false # Specifies whether the client should # back-off from (i.e. temporarily # blacklist) a failed host # (default: false). maxBackoff = 0 # Max timeout in millis that a will # remain inactive due to a previous # failure with that host (default: 0, # which effectively becomes 30000) host-selector = round_robin # The host selection strategy used # when load-balancing among hosts # (default: round_robin). # Other values are include "random" # or the FQCN of a custom class # that implements # LoadBalancingRpcClient$HostSelector batch-size = 100 # Must be >=1 (default: 100) connect-timeout = 20000 # Must be >=1000 (default: 20000) request-timeout = 20000 # Must be >=1000 (default: 20000) </pre></div> </div> </div> </div> <div class="section" id="embedded-agent"> <h4>Embedded agent<a class="headerlink" href="#embedded-agent" title="Permalink to this headline">¶</a></h4> Flume has an embedded agent api which allows users to embed an agent in their application. This agent is meant to be lightweight and as such not all sources, sinks, and channels are allowed. Specifically the source used is a special embedded source and events should be send to the source via the put, putAll methods on the EmbeddedAgent object. Only File Channel and Memory Channel are allowed as channels while Avro Sink is the only supported sink. Interceptors are also supported by the embedded agent. Note: The embedded agent has a dependency on hadoop-core.jar. Configuration of an Embedded Agent is similar to configuration of a full Agent. The following is an exhaustive list of configration options: Required properties are in bold. <table border="1" class="docutils"> <colgroup> <col width="20%" /> <col width="15%" /> <col width="65%" /> </colgroup> <thead valign="bottom"> <tr class="row-odd"><th class="head">Property Name</th> <th class="head">Default</th> <th class="head">Description</th> </tr> </thead> <tbody valign="top"> <tr class="row-even"><td>source.type</td> <td>embedded</td> <td>The only available source is the embedded source.</td> </tr> <tr class="row-odd"><td>channel.type</td> <td>–</td> <td>Either <tt class="docutils literal">memory</tt> or <tt class="docutils literal">file</tt> which correspond to MemoryChannel and FileChannel respectively.</td> </tr> <tr class="row-even"><td>channel.*</td> <td>–</td> <td>Configuration options for the channel type requested, see MemoryChannel or FileChannel user guide for an exhaustive list.</td> </tr> <tr class="row-odd"><td>sinks</td> <td>–</td> <td>List of sink names</td> </tr> <tr class="row-even"><td>sink.type</td> <td>–</td> <td>Property name must match a name in the list of sinks. Value must be <tt class="docutils literal">avro</tt></td> </tr> <tr class="row-odd"><td>sink.*</td> <td>–</td> <td>Configuration options for the sink. See AvroSink user guide for an exhaustive list, however note AvroSink requires at least hostname and port.</td> </tr> <tr class="row-even"><td>processor.type</td> <td>–</td> <td>Either <tt class="docutils literal">failover</tt> or <tt class="docutils literal">load_balance</tt> which correspond to FailoverSinksProcessor and LoadBalancingSinkProcessor respectively.</td> </tr> <tr class="row-odd"><td>processor.*</td> <td>–</td> <td>Configuration options for the sink processor selected. See FailoverSinksProcessor and LoadBalancingSinkProcessor user guide for an exhaustive list.</td> </tr> <tr class="row-even"><td>source.interceptors</td> <td>–</td> <td>Space-separated list of interceptors</td> </tr> <tr class="row-odd"><td>source.interceptors.*</td> <td>–</td> <td>Configuration options for individual interceptors specified in the source.interceptors property</td> </tr> </tbody> </table> Below is an example of how to use the agent: <div class="highlight-java"><div class="highlight"><pre>Map<String, String> properties = new HashMap<String, String>(); properties.put("channel.type", "memory"); properties.put("channel.capacity", "200"); properties.put("sinks", "sink1 sink2"); properties.put("sink1.type", "avro"); properties.put("sink2.type", "avro"); properties.put("sink1.hostname", "collector1.apache.org"); properties.put("sink1.port", "5564"); properties.put("sink2.hostname", "collector2.apache.org"); properties.put("sink2.port", "5565"); properties.put("processor.type", "load_balance"); properties.put("source.interceptors", "i1"); properties.put("source.interceptors.i1.type", "static"); properties.put("source.interceptors.i1.key", "key1"); properties.put("source.interceptors.i1.value", "value1"); EmbeddedAgent agent = new EmbeddedAgent("myagent"); agent.configure(properties); agent.start(); List<Event> events = Lists.newArrayList(); events.add(event); events.add(event); events.add(event); events.add(event); agent.putAll(events); ... agent.stop(); </pre></div> </div> </div> <div class="section" id="transaction-interface"> <h4>Transaction interface<a class="headerlink" href="#transaction-interface" title="Permalink to this headline">¶</a></h4> The <tt class="docutils literal">Transaction</tt> interface is the basis of reliability for Flume. All the major components (ie. <tt class="docutils literal">Source</tt>s, <tt class="docutils literal">Sink</tt>s and <tt class="docutils literal">Channel</tt>s) must use a Flume <tt class="docutils literal">Transaction</tt>. <div class="figure align-center"> <img alt="Transaction sequence diagram" src="_images/DevGuide_image01.png" /> </div> A <tt class="docutils literal">Transaction</tt> is implemented within a <tt class="docutils literal">Channel</tt> implementation. Each <tt class="docutils literal">Source</tt> and <tt class="docutils literal">Sink</tt> that is connected to a <tt class="docutils literal">Channel</tt> must obtain a <tt class="docutils literal">Transaction</tt> object. The <tt class="docutils literal">Source</tt>s use a <tt class="docutils literal">ChannelProcessor</tt> to manage the <tt class="docutils literal">Transaction</tt>s, the <tt class="docutils literal">Sink</tt>s manage them explicitly via their configured <tt class="docutils literal">Channel</tt>. The operation to stage an <tt class="docutils literal">Event</tt> (put it into a <tt class="docutils literal">Channel</tt>) or extract an <tt class="docutils literal">Event</tt> (take it out of a <tt class="docutils literal">Channel</tt>) is done inside an active <tt class="docutils literal">Transaction</tt>. For example: <div class="highlight-java"><div class="highlight"><pre>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(); } </pre></div> </div> Here we get hold of a <tt class="docutils literal">Transaction</tt> from a <tt class="docutils literal">Channel</tt>. After <tt class="docutils literal">begin()</tt> returns, the <tt class="docutils literal">Transaction</tt> is now active/open and the <tt class="docutils literal">Event</tt> is then put into the <tt class="docutils literal">Channel</tt>. If the put is successful, then the <tt class="docutils literal">Transaction</tt> is committed and closed. </div> <div class="section" id="sink"> <h4>Sink<a class="headerlink" href="#sink" title="Permalink to this headline">¶</a></h4> The purpose of a <tt class="docutils literal">Sink</tt> to extract <tt class="docutils literal">Event</tt>s from the <tt class="docutils literal">Channel</tt> and forward them to the next Flume Agent in the flow or store them in an external repository. A <tt class="docutils literal">Sink</tt> is associated with exactly one <tt class="docutils literal">Channel</tt>s, as configured in the Flume properties file. There’s one <tt class="docutils literal">SinkRunner</tt> instance associated with every configured <tt class="docutils literal">Sink</tt>, and when the Flume framework calls <tt class="docutils literal">SinkRunner.start()</tt>, a new thread is created to drive the <tt class="docutils literal">Sink</tt> (using <tt class="docutils literal">SinkRunner.PollingRunner</tt> as the thread’s <tt class="docutils literal">Runnable</tt>). This thread manages the <tt class="docutils literal">Sink</tt>’s lifecycle. The <tt class="docutils literal">Sink</tt> needs to implement the <tt class="docutils literal">start()</tt> and <tt class="docutils literal">stop()</tt> methods that are part of the <tt class="docutils literal">LifecycleAware</tt> interface. The <tt class="docutils literal">Sink.start()</tt> method should initialize the <tt class="docutils literal">Sink</tt> and bring it to a state where it can forward the <tt class="docutils literal">Event</tt>s to its next destination. The <tt class="docutils literal">Sink.process()</tt> method should do the core processing of extracting the <tt class="docutils literal">Event</tt> from the <tt class="docutils literal">Channel</tt> and forwarding it. The <tt class="docutils literal">Sink.stop()</tt> method should do the necessary cleanup (e.g. releasing resources). The <tt class="docutils literal">Sink</tt> implementation also needs to implement the <tt class="docutils literal">Configurable</tt> interface for processing its own configuration settings. For example: <div class="highlight-java"><div class="highlight"><pre>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; } } </pre></div> </div> </div> <div class="section" id="source"> <h4>Source<a class="headerlink" href="#source" title="Permalink to this headline">¶</a></h4> The purpose of a <tt class="docutils literal">Source</tt> is to receive data from an external client and store it into the configured <tt class="docutils literal">Channel</tt>s. A <tt class="docutils literal">Source</tt> can get an instance of its own <tt class="docutils literal">ChannelProcessor</tt> to process an <tt class="docutils literal">Event</tt>, commited within a <tt class="docutils literal">Channel</tt> local transaction, in serial. In the case of an exception, required <tt class="docutils literal">Channel</tt>s will propagate the exception, all <tt class="docutils literal">Channel</tt>s will rollback their transaction, but events processed previously on other <tt class="docutils literal">Channel</tt>s will remain committed. Similar to the <tt class="docutils literal">SinkRunner.PollingRunner</tt> <tt class="docutils literal">Runnable</tt>, there’s a <tt class="docutils literal">PollingRunner</tt> <tt class="docutils literal">Runnable</tt> that executes on a thread created when the Flume framework calls <tt class="docutils literal">PollableSourceRunner.start()</tt>. Each configured <tt class="docutils literal">PollableSource</tt> is associated with its own thread that runs a <tt class="docutils literal">PollingRunner</tt>. This thread manages the <tt class="docutils literal">PollableSource</tt>’s lifecycle, such as starting and stopping. A <tt class="docutils literal">PollableSource</tt> implementation must implement the <tt class="docutils literal">start()</tt> and <tt class="docutils literal">stop()</tt> methods that are declared in the <tt class="docutils literal">LifecycleAware</tt> interface. The runner of a <tt class="docutils literal">PollableSource</tt> invokes that <tt class="docutils literal">Source</tt>‘s <tt class="docutils literal">process()</tt> method. The <tt class="docutils literal">process()</tt> method should check for new data and store it into the <tt class="docutils literal">Channel</tt> as Flume <tt class="docutils literal">Event</tt>s. Note that there are actually two types of <tt class="docutils literal">Source</tt>s. The <tt class="docutils literal">PollableSource</tt> was already mentioned. The other is the <tt class="docutils literal">EventDrivenSource</tt>. The <tt class="docutils literal">EventDrivenSource</tt>, unlike the <tt class="docutils literal">PollableSource</tt>, must have its own callback mechanism that captures the new data and stores it into the <tt class="docutils literal">Channel</tt>. The <tt class="docutils literal">EventDrivenSource</tt>s are not each driven by their own thread like the <tt class="docutils literal">PollableSource</tt>s are. Below is an example of a custom <tt class="docutils literal">PollableSource</tt>: <div class="highlight-java"><div class="highlight"><pre>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; } } </pre></div> </div> </div> <div class="section" id="channel"> <h4>Channel<a class="headerlink" href="#channel" title="Permalink to this headline">¶</a></h4> TBD </div> <div class="section" id="initializable"> <h4>Initializable<a class="headerlink" href="#initializable" title="Permalink to this headline">¶</a></h4> As of Flume 1.10.0 Sources, Sinks, and Channels may implement the Intitializable interface. Doing so allows the component to have access the materialized configuration before any of the components have been started. While this ability is quite useful when using the standard configuration, it is less useful when configuring using Spring Boot as Spring’s autowiring generally can be used to accomplish the same thing. This example shows a Sink being configured with the name of a Source. While initializing it will retrieve the Source from the configuration and save it. During event processing a new event will be sent to the Source, presumably after the event has be modified in some way. <div class="highlight-java"><div class="highlight"><pre>public class NullInitSink extends NullSink implements Initializable { private static final Logger logger = LoggerFactory.getLogger(NullInitSink.class); private String sourceName = null; private EventProcessor eventProcessor = null; private long total = 0; public NullInitSink() { super(); } @Override public void configure(Context context) { sourceName = context.getString("targetSource"); super.configure(context); } @Override public void initialize(MaterializedConfiguration configuration) { logger.debug("Locating source for event publishing"); for (Map.Entry<String, SourceRunner> entry : configuration.getSourceRunners().entrySet()) { if (entry.getKey().equals(sourceName)) { Source source = entry.getValue().getSource(); if (source instanceof EventProcessor) { eventProcessor = (EventProcessor) source; logger.debug("Found event processor {}", source.getName()); return; } } } logger.warn("No Source named {} found for republishing events.", sourceName); } @Override public Status process() throws EventDeliveryException { Status status = Status.READY; Channel channel = getChannel(); Transaction transaction = channel.getTransaction(); Event event = null; CounterGroup counterGroup = getCounterGroup(); long batchSize = getBatchSize(); long eventCounter = counterGroup.get("events.success"); try { transaction.begin(); int i = 0; for (i = 0; i < batchSize; i++) { event = channel.take(); if (event != null) { long id = Long.parseLong(new String(event.getBody())); total += id; event.getHeaders().put("Total", Long.toString(total)); eventProcessor.processEvent(event); logger.info("Null sink {} successful processed event {}", getName(), id); } else { status = Status.BACKOFF; break; } } transaction.commit(); counterGroup.addAndGet("events.success", (long) Math.min(batchSize, i)); counterGroup.incrementAndGet("transaction.success"); } catch (Exception ex) { transaction.rollback(); counterGroup.incrementAndGet("transaction.failed"); logger.error("Failed to deliver event. Exception follows.", ex); throw new EventDeliveryException("Failed to deliver event: " + event, ex); } finally { transaction.close(); } return status; } } </pre></div> </div> </div> </div> </div> </div> </div> </div> </div> <div class="sphinxsidebar"> <div class="sphinxsidebarwrapper"><h3><a href="index.html">Apache Flume</a></h3> <ul> <li class="toctree-l1"><a class="reference internal" href="getinvolved.html">How to Get Involved</a></li> <li class="toctree-l1"><a class="reference internal" href="download.html">Download</a></li> <li class="toctree-l1"><a class="reference internal" href="security.html">Apache Flume Security Vulnerabilities</a></li> <li class="toctree-l1"><a class="reference internal" href="documentation.html">Documentation</a></li> <li class="toctree-l1"><a class="reference internal" href="releases/index.html">Releases</a></li> <li class="toctree-l1"><a class="reference internal" href="mailinglists.html">Mailing lists</a></li> <li class="toctree-l1"><a class="reference internal" href="team.html">Team</a></li> <li class="toctree-l1"><a class="reference internal" href="source.html">Source Repository</a></li> <li class="toctree-l1"><a class="reference internal" href="testing.html">Testing</a></li> <li class="toctree-l1"><a class="reference internal" href="license.html">Apache License</a></li> <li class="toctree-l1"><a class="reference internal" href="subprojects.html">Sub Projects</a></li> </ul> <h3>Resources</h3> <ul class="this-page-menu"> <li><a href="https://issues.apache.org/jira/browse/FLUME">Flume Issue Tracking (Jira)</a></li> <li><a href="https://cwiki.apache.org/confluence/display/FLUME">Flume Wiki</a></li> <li><a href="https://cwiki.apache.org/confluence/display/FLUME/Getting+Started">Getting Started Guide</a></li> </ul> <h3>Apache</h3> <ul class="this-page-menu"> <li><a href="https://www.apache.org">Home</a></li> <li><a href="https://www.apache.org/foundation/sponsorship.html">Sponsorship</a></li> <li><a href="https://www.apache.org/licenses">Licenses</a> </li> <li><a href="https://www.apache.org/foundation/thanks.html">Thanks</a></li> <li><a href="https://www.apachecon.com">Conferences</a></li> <li><a href="https://www.apache.org/security/">Security</a></li> <li><a href="https://privacy.apache.org/policies/privacy-policy-public.html">Data Privacy</a></li> </ul> <h3><a href="index.html">This Page</a></h3> <ul> <li><a class="reference internal" href="#">Flume 1.11.0 Developer Guide</a><ul> <li><a class="reference internal" href="#introduction">Introduction</a><ul> <li><a class="reference internal" href="#overview">Overview</a></li> <li><a class="reference internal" href="#architecture">Architecture</a><ul> <li><a class="reference internal" href="#data-flow-model">Data flow model</a></li> <li><a class="reference internal" href="#reliability">Reliability</a></li> </ul> </li> <li><a class="reference internal" href="#building-flume">Building Flume</a><ul> <li><a class="reference internal" href="#getting-the-source">Getting the source</a></li> <li><a class="reference internal" href="#compile-test-flume">Compile/test Flume</a></li> <li><a class="reference internal" href="#updating-protocol-buffer-version">Updating Protocol Buffer Version</a></li> </ul> </li> <li><a class="reference internal" href="#developing-custom-components">Developing custom components</a><ul> <li><a class="reference internal" href="#client">Client</a><ul> <li><a class="reference internal" href="#client-sdk">Client SDK</a></li> <li><a class="reference internal" href="#rpc-client-interface">RPC client interface</a></li> <li><a class="reference internal" href="#rpc-clients-avro-and-thrift">RPC clients - Avro and Thrift</a></li> <li><a class="reference internal" href="#secure-rpc-client-thrift">Secure RPC client - Thrift</a></li> <li><a class="reference internal" href="#failover-client">Failover Client</a></li> <li><a class="reference internal" href="#loadbalancing-rpc-client">LoadBalancing RPC client</a></li> </ul> </li> <li><a class="reference internal" href="#embedded-agent">Embedded agent</a></li> <li><a class="reference internal" href="#transaction-interface">Transaction interface</a></li> <li><a class="reference internal" href="#sink">Sink</a></li> <li><a class="reference internal" href="#source">Source</a></li> <li><a class="reference internal" href="#channel">Channel</a></li> <li><a class="reference internal" href="#initializable">Initializable</a></li> </ul> </li> </ul> </li> </ul> </li> </ul> </div> </div> <div class="clearer"></div> </div> <div class="footer"> © Copyright 2009-2023 The Apache Software Foundation. Apache Flume, Flume, Apache, the Apache feather logo, and the Apache Flume project logo are trademarks of The Apache Software Foundation.. </div> </body> </html>

content/FlumeDeveloperGuide.html (958 lines of code) (raw):