Avro Source

Flume’s primary RPC source is the Avro Source. The Avro Source is designed to be a highly scalable RPC server that accepts data into a Flume agent, from another Flume agent’s Avro Sink or from a client application that uses Flume’s SDK to send data. The Avro Source together with the Avro Sink represents Flume’s internal communication mechanism (between Flume agents). With the scalability of the Avro Source combined with the channels that act as a buffer, Flume agents can handle significant load spikes.

Flume’s Avro Source uses the Netty-Avro inter-process communication (IPC) protocol to communicate. As a result, it is possible to send data to the Avro Source from Java or JVM languages. To send events from your application to an agent with an Avro Source, you can make use of the Flume SDK (see “Flume Client SDK”) or the embedded agent (see “Embedded Agent”).

An Avro Source can be configured to accept compressed events from an Avro Sink that is configured to output them. It can also be configured to make sure that any clients or sinks sending data to it encrypt the data using SSL. An Avro Source’s configuration parameters are detailed in Table 3-3.

Configuring the Avro Source in the simplest way requires a minimal set of parameters. Minimally, the source requires two mandatory parameters other than the type parameter itself: bind and port. These two parameters define the socket address that the source uses. If there are multiple network interfaces, the Avro Source can bind to one or all of them. To bind to just one of the interfaces, simply set the IP address/domain address of that interface as the value of the bind parameter. To bind to all interfaces, use 0.0.0.0 as the value of the bind parameter. The port parameter defines the port number the source should listen on, for the configured bind address(es).

The Avro Source uses a Netty server [netty] to serve incoming requests. The Netty server uses Java’s nonblocking I/O (NIO) [nio], which allows it to be highly performant while using a relatively small number of threads to handle the requests. The Avro Source allows the user to configure the maximum number of threads that the source should use to handle incoming data using the threads parameter. This allows the user to keep a check on the resources consumed by the source. Though there is no theoretical maximum on the number of threads, the actual number is limited by the JVM, the OS, and the hardware.

If the Avro Sink(s) or RPC clients sending data to Flume are configured to use SSL to send the events to the Avro Source, the Avro Source must be configured with the SSL-related parameters. The ssl parameter must be set to true and the keystore and keystore-password parameters must be set. The keystore parameter must point to a valid keystore file, and keystore-password is the password that is to be used to open the keystore.

The keystore-type parameter is optional and can be set to an alternate keystore type, if needed [keystore-type]. The cryptographic algorithm used is defined by ssl.KeyManagerFactory.algorithm in the Java security properties file. If this parameter is not set in the Java security properties file, then the SunX509 algorithm is used. More details about the Java security properties file can be found in the Java Security Guide [java-security].

Avro Sinks and Flume’s RPC clients can be configured to compress data before sending it to the Avro Source. This is especially useful if the data is sent over a WAN, between data centers, etc. to reduce bandwidth usage. Currently, the Avro Source only supports zlib compression for RPC. To enable the Avro Source to receive data in compressed form, set the compression-type parameter to deflate. If this parameter is not set or is set to none, Flume will not attempt to decompress the data; this might cause the events to get backlogged at the previous hop, since the source will not be able to parse the compressed data and will return an error to the previous hop, causing that hop to retry forever.

Here is an example of an Avro Source configured with SSL and compression. To disable SSL, simply remove the ssl parameter, and the remaining SSL-related parameters will be ignored:

agent.sources = avroSrc
agent.channels = memChannel

agent.sources.avroSrc.type = avro
agent.sources.avroSrc.channels = memChannel

# Bind to all interfaces
agent.sources.avroSrc.bind = 0.0.0.0
agent.sources.avroSrc.port = 4353

# Removing the next line will disable SSL
agent.sources.avroSrc.ssl = true
agent.sources.avroSrc.keystore = /tmp/keystore.jks
agent.sources.avroSrc.keystore-password = UsingFlume
agent.sources.avroSrc.keystore-type = jks

agent.sources.avroSrc.compression-type = deflate

# Initializes a memory channel with default configuration
agent.channels.memChannel.type = memory

An Avro Sink that writes to this source would have a configuration similar to the following:

agent.channels = avroSinkChannel
agent.sinks = avroSink

agent.channels.avroSinkChannel.type = memory

agent.sinks.avroSink.type = avro
agent.sinks.avroSink.channel = memory
agent.sinks.avroSink.hostname = avrosrchost.example.com
agent.sinks.avroSink.port = 4353

# SSL properties
agent.sinks.avroSink.ssl = true
agent.sinks.avroSink.trust-all-certs = true
agent.sinks.avroSink.truststore = /path/to/keystore
agent.sinks.avroSink.truststore-password = UsingFlume
agent.sinks.avroSink.truststore-type = JKS

agent.sources.avroSink.compression-type = deflate

Thrift Source

As mentioned in “Avro Source”, Flume’s use of Avro’s Java-specific RPC mechanism makes the Avro Source unable to accept data from non-JVM languages. As Flume became more popular, this use case had to be addressed. Therefore, Apache Thrift RPC support [thrift_ch3] was added to Flume. Thrift is a top-level project at the Apache Software Foundation that enables cross-language communication, which is extremely popular. The Thrift Sink–Thrift Source combination in Flume is designed to work pretty much exactly like the Avro Sink–Avro Source combination. Flume also has a Java Thrift RPC client that is part of the Flume SDK. The Thrift Source, in the simplest terms, is a multithreaded high-performance Thrift server. The Thrift interface definition language (IDL) that Flume uses is shown here:

namespace java org.apache.flume.thrift

struct ThriftFlumeEvent {
  1: required map <string, string> headers,
  2: required binary body,
}

enum Status {
  OK,
  FAILED,
  ERROR,
  UNKNOWN
}

service ThriftSourceProtocol {
  Status append(1: ThriftFlumeEvent event),
  Status appendBatch(1: list<ThriftFlumeEvent> events),
}

This IDL can be used to generate Thrift clients in any language that Thrift supports. The generated code can then be used to send data to Flume’s Thrift Source.

The configuration of the Thrift Source is extremely simple and mimics that of the Avro Source (see Table 3-4).

The bind parameter specifies the hostname/IP address of the interface to bind to; use 0.0.0.0 to bind to all interfaces. The port parameter specifies the port to use—this is the port clients would use to send events to this source. These are both mandatory parameters.

The threads parameter for a Thrift Source works in a slightly different way than for the Avro Source. Flume (as of version 1.4.0), by default, is built against and includes Thrift version 0.7.0. This was meant to support clients (programs written to use the Flume SDK) that would also write to HBase version 0.92 (or older) from the same process. If there is no requirement to support this version of HBase, then it is recommended that the Thrift version that comes with Flume be replaced with a newer version, though due to incompatibilities in the Thrift-generated code, Flume may also need to be recompiled against the newer version.

When using a version of Thrift lower than 0.8.0, Flume uses Thrift’s TThreadPoolServer, which uses one thread per client connected, and the threads parameter controls the maximum number of threads that the source will create, thus indirectly controlling the number of clients that can connect to the agent. It is recommended to not set this parameter in this case. If a newer version of Thrift is being used, then Flume uses Thrift’s TThreadedSelectorServer, which uses Java’s nonblocking I/O and therefore can support more clients than there are threads available. In this case, the threads parameter works just like the Avro Source’s threads parameter and can be used to keep the resource utilization under control.

The Thrift Source, unlike the Avro Source, does not currently support compression or SSL. A Thrift Source should therefore be used only to push data into Flume from systems that are written in non-JVM languages, or if the application that is writing the data already uses Thrift for other purposes. For Flume agent–to–Flume agent communication, it is recommended that the Avro Sink–Avro Source pair be used.

The following is an example of configuration of a Thrift Source:

agent.sources = thriftSrc
agent.channels = memChannel

agent.sources.thriftSrc.type = thrift
agent.sources.thriftSrc.channels = memChannel

# Bind to all interfaces
agent.sources.thriftSrc.bind = 0.0.0.0
agent.sources.thriftSrc.port = 4564

# Initializes a memory channel with default configuration
agent.channels.memChannel.type = memory

A Thrift Sink that writes to this source would have a configuration similar to the following:

agent.channels = memChannel
agent.sinks = thriftSink

agent.channels.memChannel.type = memory

agent.sinks.thriftSink.type = thrift
agent.sinks.thriftSink.channel = memory
agent.sinks.thriftSink.hostname = thriftsrchost.example.com
agent.sinks.thriftSink.port = 4564

Failure Handling in RPC Sources

Failure handling in both the Avro and Thrift Sources is a bit tricky. This is because the RPC sources are invoked by a client or sink on the other side of a network link, though it looks like a local method call. In all cases where the RPC sources cannot start due to some permanent error, like being unable to bind to the port, the source will throw an exception when it tries to start. Since Flume’s configuration system will retry every few seconds to restart the component, since it was successfully configured, the source will start up if the condition causing the error no longer exists—for example, if the other process that was bound to the port was killed or released the port.

The trickier part, though, is with respect to the code that actually receives the data and writes the events to the channel. If even one of the channels the source is configured to write to throws a ChannelException due to the channel being full, or if the transaction is too large, the source returns a failure status to the client or sink that called it and expects it to retry. Since RPC sources receives data via threads owned by a thread pool, exceptions would simply cause the thread to die.

In all such cases, the real indication of failure is only in the log files where these exceptions are logged. Sometimes these exceptions may indicate a major problem, like the process running out of resources (as with an OutOfMemoryError). Therefore, it is important to monitor the logs generated by the Flume agent to ensure that things are running smoothly. ChannelExceptions being thrown too often can mean that the channels are too underallocated for the rate of writes, or that the sinks are not clearing the data from the channels fast enough. Increasing the number of sinks can help if too few sinks are reading the data, but if the eventual destination itself cannot handle the load, the capacity needs to be rethought. In all cases, errors may cause duplicates but never actually cause data loss, since events are removed from the channel if and only if the data is actually successfully written out to the next hop.

Writing Handlers for the HTTP Source*

It is easy for the user to develop and plug in a handler to convert the data received into Flume events. This allows the HTTP Source to accept data from clients in any format that can be processed by the handler. The HTTP Source handler is a class that inherits a very simple interface, HTTPSourceHandler:

package org.apache.flume.source.http;
public interface HTTPSourceHandler extends Configurable {
  public List<Event> getEvents(HttpServletRequest request) throws
          HTTPBadRequestException, Exception;
}

The handler interface is extremely simple and has only one method, getEvents, which accepts the HTTPServletRequest sent by the client and returns a list of Flume events. Even though this handler interface is simple, it can essentially do any arbitrary processing to convert the input data from the HTTPServletRequest into Flume events. The amount of processing should be limited, though, or the client sending data to this source might get timed out. The handler is configurable through Flume’s standard configuration mechanism. Since the HTTP Source always uses exactly one transaction per request whatever the handler is, the sender has to be careful to send only as many events as the channels support.

The handler is responsible for making sure that the configuration parameters passed to it are valid. The HTTP Source will instantiate and configure the handler on startup. Since the HTTP Source propagates any Exception thrown by the handler to the configuration system, the handler must verify the parameters and apply the parameters in the configure method. The parameters are passed in to the configure method via a Context instance. Context instances are simply key-value pairs containing various configuration keys and their values. If the configuration passed in is invalid and cannot be applied successfully, the HTTP Source rethrows the exception thrown by the handler to the configuration system, which in turn disables the HTTP Source and removes it from the agent.

While processing incoming data, the HTTP Source handles exceptions thrown by the handler by returning a failure to the client. The HTTP Source expects the handler to throw an HTTPBadRequestException if the incoming data was malformed and cannot be converted into Flume events. This operation must be idempotent, and the handler must throw the same exception for the same input every time. If an HTTPBadRequestException is thrown by the handler, the HTTP Source returns HTTP error code 400, to inform the client that the request was malformed. If the handler throws any other exception, the source returns HTTP error code 500, to inform the client that there was an internal error in the HTTP Source. It is then up to the client to decide how to retry in such a case. If one of the channels that the source is writing to throws a channel exception, the source returns error code 503, to signal that the channel is temporarily at capacity and the client should retry later.

The JAR file containing the handler (or the handler’s .class file) and all its dependencies should be added to the Flume classpath via the plugins.d mechanism discussed in “Deploying Custom Code”.

If no handler is specified in the configuration, the HTTP Source uses a handler that comes bundled with Flume, which can handle JSON formatted events. Each request can contain several events represented as an array, though the channel(s) the source writes to might have a limited transaction capacity. The handler accepts JSON-formatted data in the UTF-8, UTF-16, or UTF-32 charset, and converts it into a list of events with the body serialized in the charset of the original HTTP request. The format that the handler accepts is shown here:

[{
    "headers" : {
		"event1Header1" : "event1Value1",
		"event1header2" : "event1Value2"
	},
	"body" : "This is the body of the first event."

},
{
	"headers" : {
		"event2Header1" : "event2Value1",
		"event2Header2" : "event2Value2"
	},
	"body" : "This is the body of the second event"
}]

HTTPSourceXMLHandler, shown in Example 3-5, is another example of a handler that works with the HTTP Source. This handler converts XML-formatted data into Flume events. The handler is pretty simple and expects the data to be in the XML format shown in Example 3-4. The format expected by this handler is pretty simple. Only data in between <events> and </events> is processed. Each event is expected to be between <event> and </event> tags. The only thing that limits the number of events per request is the transaction capacity of the channel(s) the source writes to. Each event can have one or more sections, enclosed by <headers> and </headers> tags. Each header is denoted by a tag whose name is used as the header name; the value in between the opening header name tag and the closing tag is used as its value. The body is enclosed between <body> and </body> tags.

<events>
  <!-- This can contain as many events
  as the channel can support in a transaction -->

  <event>
    <headers>
      <header1>value1</header1>
      <header2>value2</header2>
    </headers>

    <body>This is a test.
        This input should show up in an event.
    </body>
  </event>

  <event>
    <!-- There can be zero or more headers sections.
     They are merged together, so each header name
     must be unique even between sections. -->
    <headers>
      <event2Header1>event2Value1</event2Header1>
    </headers>

    <!-- Each event can have only one body -->

    <body>This is the 2nd event.</body>

    <headers>
      <event2Header2>event2Value2</event2Header2>
    </headers>
  </event>
</events>

The handler parses the XML-formatted events into Flume events and returns them to the HTTP Source, which in turn writes them to the channel(s). While parsing the events, the handler makes sure that each event has at least a header and a body. If not, the handler throws an HTTPBadRequestException to inform the client that the incoming data was malformed. It can be configured to insert a timestamp into the Flume event headers, which the HDFS Sink can use for bucketing of events.

package usingflume.ch03;

/**
 * A handler for the HTTP Source that accepts XML-formatted data.
 * Each event can contain multiple header nodes,
 * but exactly one body node. If there is
 * more than one body tag, the first one in the event is picked up.
 */
public class HTTPSourceXMLHandler implements HTTPSourceHandler {

  private final String ROOT = "events";
  private final String EVENT_TAG = "event";
  private final String HEADERS_TAG = "headers";
  private final String BODY_TAG = "body";

  private final String CONF_INSERT_TIMESTAMP = "insertTimestamp";
  private final String TIMESTAMP_HEADER = "timestamp";

  private final DocumentBuilderFactory documentBuilderFactory
    = DocumentBuilderFactory.newInstance();

  private final ThreadLocal<DocumentBuilder> docBuilder
    = new ThreadLocal<DocumentBuilder>();

  private boolean insertTimestamp;

  @Override
  public List<Event> getEvents(HttpServletRequest
    httpServletRequest) throws HTTPBadRequestException, Exception {
    if (docBuilder.get() == null) {
      docBuilder.set(documentBuilderFactory.newDocumentBuilder());
    }
    Document doc;
    final List<Event> events;
    try {
      doc = docBuilder.get().parse(
        httpServletRequest.getInputStream());
      Element root = doc.getDocumentElement();
      root.normalize();

      // Verify that the root element is "events"
      Preconditions.checkState(
        ROOT.equalsIgnoreCase(root.getTagName()));

      NodeList nodes = root.getElementsByTagName(EVENT_TAG);
      int eventCount = nodes.getLength();
      events = new ArrayList<Event>(eventCount);
      for (int i = 0; i < eventCount; i++) {
        Element event = (Element) nodes.item(i);
        // Get all headers. If there are multiple header sections,
        // combine them.
        NodeList headerNodes
          = event.getElementsByTagName(HEADERS_TAG);
        Map<String, String> eventHeaders
          = new HashMap<String, String>();
        for (int j = 0; j < headerNodes.getLength(); j++) {
          Node headerNode = headerNodes.item(j);
          NodeList headers = headerNode.getChildNodes();
          for (int k = 0; k < headers.getLength(); k++) {
            Node header = headers.item(k);

            // Read only element nodes
            if (header.getNodeType() != Node.ELEMENT_NODE) {
              continue;
            }
            // Make sure a header is inserted only once,
            // else the event is malformed
            Preconditions.checkState(
              !eventHeaders.containsKey(header.getNodeName()),
              "Header expected only once " + header.getNodeName());
            eventHeaders.put(
              header.getNodeName(), header.getTextContent());
          }
        }
        Node body = event.getElementsByTagName(BODY_TAG).item(0);
        if (insertTimestamp) {
          eventHeaders.put(TIMESTAMP_HEADER, String.valueOf(System
            .currentTimeMillis()));
        }
        events.add(EventBuilder.withBody(
          body.getTextContent().getBytes(
            httpServletRequest.getCharacterEncoding()),
          eventHeaders));
      }
    } catch (SAXException ex) {
      throw new HTTPBadRequestException(
        "Request could not be parsed into valid XML", ex);
    } catch (Exception ex) {
      throw new HTTPBadRequestException(
        "Request is not in expected format. " +
          "Please refer documentation for expected format.", ex);
    }
    return events;
  }

  @Override
  public void configure(Context context) {
    insertTimestamp = context.getBoolean(CONF_INSERT_TIMESTAMP,
      false);
  }
}

An example of XML handler configuration is shown in Example 3-3. This configuration instructs the handler to insert the timestamp of processing into each event. In general, any number of parameters can be passed to HTTP Source handlers in this way.

The HTTPHandler interface is part of the flume-ng-core Maven artifact, which can be added to your application by including Example 3-6 in your application’s pom.xml file’s dependency section.

    <dependency>
      <groupId>org.apache.flume</groupId>
      <artifactId>flume-ng-core</artifactId>
      <version>1.5.0</version>
    </dependency>

Use the plugins.d framework shown in “Deploying Custom Code” to deploy custom HTTP handlers to Flume agents.

Reading Custom Formats Using Deserializers*

The source deserializes data in the files in the directory using a pluggable deserializer, allowing the source to read and interpret the data from these files into events in different ways. For example, a deserializer that “understands” Avro could read Avro container formatted files and convert each Avro message to a Flume event, or several lines could be read at once, until an entire stack trace is read and then converted to Flume events. Once all data in a file is read, this source can either delete or rename the file with a new extension so that the same file is not processed again.

To use a custom deserializer, set the value of the deserializer parameter to an implementation of EventDeserializer$Builder that can build the EventDeserializer implementation to use. The deserializer can be configured by passing in parameters with the deserializer. prefix. Text-based deserializers can call the readChar method to read a character. The way a character is represented differs by character set. To tell the source what character set to use, set inputCharset to the name of the character set, which by default is UTF-8.

Deserializers implement the EventDeserializer interface, and should also provide a Builder class, which must implement the EventDeserializer$Builder interface. The Builder must have a public, no-argument constructor that the Flume framework can use to instantiate the builder. The Builder’s build method must create and return a fully configured instance of the deserializer.

A Context instance and an instance of ResettableInputStream are passed to this method. The Context instance can be used to configure the deserializer. The deserializer is expected to deserialize events from the input stream. ResettableInputStream is an interface that is meant to read data from the stream, but also gives the ability to roll back to a previous location in the stream. An instance of the ResettableInputStream class guarantees that a reset call will reset the reads from this stream to the position in the stream at which the last mark call happened, regardless of how many bytes were read from the stream using the read or readChar methods after the last mark call. This allows the Spooling Directory Source to reread events if writes to a channel failed and the events could not be written. The deserializer can use this functionality in its own mark and reset methods to ensure it rolls back to the correct location within the stream.

The deserializer implements two other methods that are called by the source to read events from the stream—the readEvent and readEvents methods. The readEvent method must return exactly one event from the stream, while the readEvents method takes an argument that is the maximum number of events it must read from the stream.

Example 3-9 shows a deserializer that deserializes messages serialized as Protocol Buffer (Protobuf) messages based on the format shown in Example 3-8. Each Protobuf message is written to the file after its length is written to the file as a 4-byte integer.

option java_package = "usingflume.ch03";
option java_outer_classname = "UsingFlumeEvent";

message Event {
  repeated Header header = 1;
  required bytes body = 2;
}

message Header {
  required string key = 1;
  required string val = 2;
}

package usingflume.ch03;

public class ProtobufDeserializer implements EventDeserializer {
  private final ResettableInputStream stream;
  private boolean isOpen;

  private ProtobufDeserializer(ResettableInputStream stream) {
    // No configuration to do, so ignore the context.
    this.stream = stream;
    isOpen = true;
  }

  @Override
  public Event readEvent() throws IOException {
    throwIfClosed();
    // To not create an array each time or copy arrays multiple times,
    // read the data to an array that backs byte buffers,
    // then wrap that array in a stream and pass it to the Protobuf
    // parseDelimitedFrom method.
    // The format is expected to be:
    // <length of message> - int
    // <protobuf message (written using writeTo (not delimited)>
    // We assume here that the file is well-formed and the length
    // or the
    // message are not partially cut off.
    byte[] sz = new byte[4];
    if (stream.read(sz, 0, 4) != -1) {
      int length = ByteBuffer.wrap(sz).getInt();
      byte[] data = new byte[length];
      stream.read(data, 0, data.length);
      UsingFlumeEvent.Event protoEvent =
        UsingFlumeEvent.Event.parseFrom(new ByteArrayInputStream(data));
      List<UsingFlumeEvent.Header> headerList
        = protoEvent.getHeaderList();
      Map<String, String> headers = new HashMap<String, String>(
        headerList.size());
      for (UsingFlumeEvent.Header hdr : headerList) {
        headers.put(hdr.getKey(), hdr.getKey());
      }
      return EventBuilder.withBody(protoEvent.getBody().toByteArray(), headers);
    }
    return null;
  }

  @Override
  public List<Event> readEvents(int count) throws IOException {
    throwIfClosed();
    List<Event> events = new ArrayList<Event>(count);
    for (int i = 0; i < count; i++) {
      Event e = readEvent();
      if (e == null) {
        break;
      }
      events.add(e);
    }
    return events;
  }

  @Override
  public void mark() throws IOException {
    throwIfClosed();
    stream.mark();
  }

  @Override
  public void reset() throws IOException {
    throwIfClosed();
    stream.reset();
  }

  @Override
  public void close() throws IOException {
    isOpen = false;
    stream.close();
  }

  private void throwIfClosed() {
    Preconditions.checkState(isOpen, "Serializer is closed!");
  }

  public static class ProtobufDeserializerBuilder implements Builder {

    @Override
    public EventDeserializer build(Context context,
      ResettableInputStream resettableInputStream) {
      // The serializer does not need any configuration,
      // so ignore the Context instance. If some configuration has
      // to be passed to the serializer, this Context instance can be used.
      return new ProtobufDeserializer(resettableInputStream);
    }
  }
}

This ProtobufDeserializer class reads Protobuf-serialized events from the file and converts them into Flume events in the readEvent method, returning null when no more events are available to be read. If no events could be read by the readEvents method, an empty list is returned, as is mandated by the EventDeserializer interface.

Since the file is immutable, once we reach a stage where no more events are available in the file, this means all the events have been read out from the file, at which point the source closes the deserializer by calling the close method. If the serializer is maintaining any internal state or has some cleanup to do, this method is expected to do that. In this case, we simply close the stream. The mark and reset methods simply check that the deserializer is open and forward the calls to the stream. This specific implementation of the serializer does not need any configuration, but deserializers can receive configuration via the Context instance passed to the builder, which in turn can be passed through a constructor to the deserializer instance. Example 3-7 showed a Spooling Directory Source configured to use ProtobufDeserializer.

Flume comes bundled with a handful of deserializers. The default deserializer is the LineDeserializer [line-deserializer]. This is an example of a deserializer that accepts configuration. The line deserializer is enabled if no deserializer is set for the Spooling Directory Source or if the value of the deserializer parameter is set to line. The line deserializer reads the file line by line, converting each line to an event, based on a configurable character set (UTF-8 by default). Table 3-7 lists the configuration parameters.

Another deserializer that comes bundled with Flume is the AvroEventDeserializer. To use this deserializer, set the deserializer parameter to avro. The Avro deserializer can read Avro container files and send data out as Avro-formatted events. There is only one configuration parameter for this deserializer, as described in Table 3-8.

To interpret the data, it is important to know the schema used. Though this information is contained in the file itself, it is important to keep the schema with each event so that it is possible to read the data from the event. So, this deserializer supports inserting the Avro schema in the headers or simply putting the 64-bit Rabin fingerprint of the schema (as specified in the Avro specification [schema-fp]) in the headers, which can later be used to look up a schema registry that is indexed on the schema fingerprint. (Imagine a situation where there is Avro-formatted data that is going to be in one of several known schemas, which is mostly the case. Using this fingerprint in the headers allows the user to identify which schema should be used to read the message, and thus this can be used by a serializer while writing the data.) To set the fingerprint in the event headers, set the schemaType parameter to flume.avro.schema.hash. The schema fingerprint is written to a header with the key flume.avro.schema.hash.

If the entire JSON-ified schema should be written to every event, set schemaType to flume.avro.schema.literal. In this case, the entire schema is written with the key flume.avro.schema.literal. Writing the schema in every event’s header is pretty inefficient since it increases the event size, especially if there are a limited number of schema types.

To read files from a directory as a binary large object (BLOB) [blob], the blob deserializer can be used. The FQCN of the blob deserializer is org.apache.flume. sink.solr.morphline.BlobDeserializer. This deserializer takes the maximum size of a blob it should accept, and for each file attempts to read data from the file in blobs of that many bytes. This serializer takes only one configuration parameter, maxBlobLength, which is the maximum size, in bytes, of each blob. If a file is larger than this, the file is split up into several blobs, each with a size less than equal to the configured maximum. This deserializer buffers all blobs in a batch in memory, so the batch size and maximum blob size should be configured to ensure that the serializer does not end up using more memory than expected.

Since the Spooling Directory Source is also a part of the flume-ng-core artifact, make sure you add the flume-ng-core artifact to your serializer’s pom.xml file, as shown in Example 3-6. Custom deserializers can be deployed to a Flume agent using the plugins.d framework shown in “Deploying Custom Code”.

Spooling Directory Source Performance

A Spooling Directory Source is I/O-bound. To avoid complicating deserializer implementations, the source was specifically designed to be single-threaded. This means that it is possible that the performance could be improved by using more threads to read the data and use more of the available CPUs. One way of improving performance of files being read is to write the files alternately to different directories and have one Spooling Directory Source process each of the directories (and write to the same channel, if all the data is going to the same destination). This means more threads read data from the disk and more of the CPUs can be utilized for deserialization.

Converting JMS Messages into Flume Events*

The JMS Source can be configured to use custom code to convert JMS messages into Flume events, much like the HTTP Source’s handler. This makes the JMS Source extremely flexible, and allows the user to parse data in proprietary formats in the JMS message. Flume comes packaged with a JMS Message Converter that supports default JMS formats. The JMS Source can pass configuration into the message converter, just like the HTTP Source. This configuration can be used to set up any initial configuration or state required for the converter. All of the JMS transaction handling is done by the source; the converter need not worry about any of that.

The default JMS Message Converter that comes packaged with Flume converts individual JMS messages into Flume events based on their format and content. A JMS ByteMessage is simply read, and the bytes read from the message are placed into the body of a Flume event. For a JMS TextMessage, the converter encodes the text into a byte array and places it into the body of a Flume event. The encoding to be used is configurable through the source configuration. If an ObjectMessage is read off the JMS queue, the converter wraps the object in an ObjectOutputStream and writes it out to a ByteArrayOutputStream. The bytes are then read from this stream and set as the Flume event’s body. In most cases, this is what users need.

Sometimes, though, it might make more sense to parse out any schemas in the message and convert them into a format that can be serialized at the terminal sink more easily. For example, if there are several different applications writing to the JMS queue in different serialization formats like JSON or XML, then a converter can be written to convert these into a unified format that can be parsed more easily at the terminal sink.

Converters must implement the JMSMessageConverter interface, which is shown in Example 3-12. The JMS Source instantiates a JMSMessageConverter$Builder class, and then passes the configuration via a Context instance to the Builder’s build method, which is expected to return a fully configured converter instance.

package org.apache.flume.source.jms;
public interface JMSMessageConverter {
  public List<Event> convert(Message message)
	   throws JMSException;
  /**
   * Implementors of JMSMessageConverter must either provide
   * a suitable builder or implement the Configurable interface.
   */
  public interface Builder {
    public JMSMessageConverter build(Context context);
  }
}

Example 3-13 shows an example of a JMS Message Converter that reads the JMS messages formatted as JSON and converts them into Flume events. Converters can implement Configurable to accept configuration, and the JMS Source passes any parameters specified with the converter. prefix to the converter. This converter accepts a configuration that tells the converter which charset to use to convert the JSON events to Flume events.

package usingflume.ch03;

public class JsonMessageConverter implements JMSMessageConverter,
  Configurable {

  private static final Logger LOGGER =
    LoggerFactory.getLogger(JsonMessageConverter.class);
  private final Type listType
    = new TypeToken<List<JSONEvent>>() {
  }.getType();
  private final Gson gson
    = new GsonBuilder().disableHtmlEscaping().create();
  private String charset = "UTF-8";

  @Override
  public List<Event> convert(javax.jms.Message message)
    throws JMSException {

    Preconditions.checkState(message instanceof TextMessage,
      "Expected a text message, but the message received " +
        "was not Text");
    List<JSONEvent> events =
      gson.fromJson(((TextMessage) message).getText(), listType);
    return convertToNormalEvents(events);
  }

  private List<Event> convertToNormalEvents(List<JSONEvent> events) {
    List<Event> newEvents = new ArrayList<Event>(events.size());
    for(JSONEvent e : events) {
      e.setCharset(charset);
      newEvents.add(EventBuilder.withBody(e.getBody(),
        e.getHeaders()));
    }
    return newEvents;
  }

  @Override
  public void configure(Context context) {
    try {
    charset = context.getString("charset", "UTF-8");
    } catch (Exception ex) {
      LOGGER.warn("Charset not found. Using UTF-8 instead", ex);
      charset = "UTF-8";
    }

  }
}

The converter shown in Example 3-13 simply takes a message that it expects to be a JMS TextMessage. This message is expected to be in JSON format converted into a list of Flume JSON events. Once the JSON events are created, a list of “simple” Flume events are created to avoid the additional overhead of converting the string to a byte array each time it is read. To configure the JMS Source with this converter, Example 3-11 can be used.

To deploy a custom converter, make sure the JAR file containing the class is deployed in the plugins.d directory and the converter.type parameter specifies the FQCN of the class that is to be used. While writing your deserializer, include the flume-jms-source artifact in your application’s pom.xml file:

    <dependency>
      <groupId>org.apache.flume.flume-ng-sinks</groupId>
      <artifactId>flume-hdfs-sink</artifactId>
      <version>1.5.0</version>
    </dependency>

Custom converters can be installed to the plugins.d directory as explained in “Deploying Custom Code”.

Event-Driven and Pollable Sources

Each source is run in its own thread, called a SourceRunner. The source runner runs a single thread that operates on the source. Flume has two types of sources: event-driven sources and pollable sources. Based on the type of source, the Flume framework creates an EventDrivenSourceRunner or a PollableSourceRunner to run the source.

package usingflume.ch03;

public class StockTickerSource extends AbstractPollableSource {

  private static final String CONF_TICKERS = "tickers";
  private static final String CONF_REFRESH_INTERVAL = "refreshInterval";
  private static final int DEFAULT_REFRESH_INTERVAL = 10;

  private int refreshInterval = DEFAULT_REFRESH_INTERVAL;

  private final List<String> tickers = new ArrayList<String>();
  private final QuoteProvider server = new RandomQuoteProvider();

  private volatile long lastPoll = 0;

  @Override
  protected Status doProcess() throws EventDeliveryException {
    Status status = Status.BACKOFF;
    if(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - lastPoll) >
      refreshInterval) {
      final List<Event> events = new ArrayList<Event>(tickers.size());
      Map<String, Float> prices = server.getQuote(tickers);
      lastPoll = System.currentTimeMillis();
      // Convert each price into ticker = price form in UTF-8 as event body
      for(Map.Entry<String, Float> e: prices.entrySet()) {
        StringBuilder builder = new StringBuilder(e.getKey());
        builder.append(" = ").append(e.getValue());
        events.add(EventBuilder.withBody(builder.toString().getBytes(Charsets
          .UTF_8)));
      }
      getChannelProcessor().processEventBatch(events);
      status = Status.READY;
    }
    return status;
  }

  @Override
  protected void doConfigure(Context context) throws FlumeException {
    refreshInterval = context.getInteger(CONF_REFRESH_INTERVAL,
      DEFAULT_REFRESH_INTERVAL);
    String tickersString = context.getString(CONF_TICKERS);
    Preconditions.checkArgument(tickersString != null && !tickersString
      .isEmpty(), "A list of tickers must be specified");
    tickers.addAll(Arrays.asList(tickersString.split("\s+")));
  }

  @Override
  protected void doStart() throws FlumeException {
    server.start();
  }

  @Override
  protected void doStop() throws FlumeException {
    server.stop();
  }
}

@namespace("usingflume.ch03")

protocol FlumeCreditCardAuth {

  record CreditCardTransaction {
    string cardNumber;
    string location;
    float amount;
  }

  enum Status {
    OK,
    FAILED
  }

  Status transactionsCompleted(array<CreditCardTransaction> transactions);
}

package usingflume.ch03;

public class TransactionSource extends AbstractEventDrivenSource implements
  FlumeCreditCardAuth {

  private static final String CONF_HOST = "host";
  private static final String CONF_PORT = "port";
  private static final String DELIMITER = ";";

  private String host;
  private Integer port;
  private Server srv;

  @Override
  protected void doConfigure(Context context) throws FlumeException {
    host = context.getString(CONF_HOST);
    Preconditions.checkArgument(host != null && !host.isEmpty(),
      "Host must be specified");
    port = Preconditions.checkNotNull(context.getInteger(CONF_PORT),
      "Port must be specified");
  }

  @Override
  protected void doStart() throws FlumeException {
    srv = new NettyServer(
      new SpecificResponder(FlumeCreditCardAuth.class, this),
      new InetSocketAddress(host, port));
    srv.start();
    super.start();
  }

  @Override
  protected void doStop() throws FlumeException {
    srv.close();
    try {
      srv.join();
    } catch (InterruptedException e) {
      throw new FlumeException("Interrupted while waiting for Netty Server to" +
        " shut down.", e);
    }
  }

  @Override
  public Status transactionsCompleted(List<CreditCardTransaction> transactions)
    throws AvroRemoteException {
    Status status;
    List<Event> events = new ArrayList<Event>(transactions.size());
    for (CreditCardTransaction txn : transactions) {
      StringBuilder builder = new StringBuilder();
      builder.append(txn.getCardNumber()).append(DELIMITER)
        .append(txn.getLocation()).append(DELIMITER)
        .append(txn.getAmount()).append(DELIMITER);
      events.add(EventBuilder.withBody(builder.toString().getBytes(
        Charsets.UTF_8)));
    }
    try {
      getChannelProcessor().processEventBatch(events);
      status = Status.OK;
    } catch (Exception e) {
      status = Status.FAILED;
    }
    return status;
  }
}