Apache Flume External Adapters

The TIBCO StreamBase© Adapters for Apache Flume allow StreamBase to integrate with Apache Flume as either a Flume sink or source. The sink and source run inside the Apache Flume agent process. This means that the process is not an external process; rather, it is loaded as a Flume agent in the context of the Flume system.

StreamBase provides two Flume-related adapters:

The Flume adapter installation kit includes a sample application. The sample appears in StreamBase Studio's Load StreamBase Sample dialog only after you install the Flume External Adapter kit.

It is possible to install the Flume adapter kit on machines that do not have StreamBase installed. On UNIX platforms, you must also install the StreamBase client JAR file, as described below in the Configuration section.

The adapters are also certified by Cloudera Inc. for use with the Flume services of their CDH 5 product.

The Flume Input adapter acts as a Flume sink that consumes Flume events and enqueues them to StreamBase Server. Conversely, the Output adapter acts as a Flume source which dequeues messages from StreamBase Server and emits corresponding Flume events.

Because the format of Flume events has no predefined format, flexibility was built into the adapter to convert Flume events to and from StreamBase tuples using a plug-in model to load the code responsible for the conversion process.

The adapters support Apache Flume versions 1.1.0 and above.

The Input adapter runs as a Flume sink node. This means it connects to a Flume channel and receives Flume events. As it receives such events it converts them to tuples and sends them to a StreamBase Server. Conversely, the Output adapter runs as a Flume source node, receiving tuples from the StreamBase Server and emitting Flume events on the configured Flume channel.

The Flume channel to which the adapter connects is determined by the settings in the Flume configuration file specified by the -f switch when invoking flume-ng. Please refer to the Flume documentation for details on the format of this file.

The StreamBase Server and stream to which the adapter connects as well as other adapter-specific settings are determined by the following parameters, also specified in the Flume configuration file. The supported parameters are:

Additionally, custom parameters can be specified in the Flume configuration file for use by the StreamBaseSink.Converter or StreamBaseSource.Converter implementer. These parameters are passed along in the form of a Flume context object (org.apache.flume.Context). See Event Translation below for more information.

The Input adapter must convert incoming Flume events to StreamBase tuples, and the Output adapter must perform the opposite conversion, tuples to Flume events. Since the body of Flume events carry a payload of arbitrary format, the adapter must provide total flexibility in the conversion mechanism to handle this payload. To this end, the adapters use a plug-in mechanism to perform custom conversions. For example, you can provide a custom converter to consume Flume events produced by an Apache Avro Flume source (or on the Output side to produce such events destined to an Avro Flume sink).

Custom converters for use by the Input adapter are expected to implement the com.streambase.sb.adapter.flume.StreamBaseSink.Converter interface. The definition of this interface is as follows:

/**
 * {@link StreamBaseSink} uses pluggable converters which must implement this interface.
 * An instance of the converter will be created using its default public constructor,
 * then the {@link #init(Context, Schema)} method will be called to perform initialization
 * of the instance. This call is guaranteed to occur before the first call to {@link #convert(Tuple, Event)}.
 */
public static interface Converter
{
  /**
   * This method can be used to initialize the object. It is guaranteed to be called
   * before the first call to {@link #convert(Tuple, Event)}.
   * @param context the flume {@link Context} used by this source. This can be used
   * by the converter instance to access custom parameters if desired.
   * @param schema the StreamBase {@link Schema} of the stream used by this sink.
   * This can be useful to perform schema validation, and to implement field-caching
   * strategies for more efficient conversions. 
   * @throws StreamBaseException
   */
  public void init(Context context, Schema schema) throws StreamBaseException;
        
  /**
   * Converts the Flume {@link Event} to a StreamBase {@link Tuple}.
   * @param event the Flume event to convert.
   * @param tuple the StreamBase tuple into which to place the event's contents.
   * @return true if the conversion was successful, false otherwise.
   * @throws StreamBaseException
   */
  public boolean convert(Event event, Tuple tuple) throws StreamBaseException;       
}

Similarly, custom converters for use by the Output adapter are expected to implement the com.streambase.sb.adapter.flume.StreamBaseSource.Converter interface. The definition of this interface is as follows:

/**
 * {@link StreamBaseSource} uses pluggable converters which must implement this interface.
 * An instance of the converter will be created using its default public constructor,
 * then the {@link #init(Context, Schema)} method will be called to perform initialization
 * of the instance. This call is guaranteed to occur before the first call to {@link #convert(Tuple, Event)}.
 */
public static interface Converter
{
  /**
   * This method can be used to initialize the object. It is guaranteed to be called
   * before the first call to {@link #convert(Tuple, Event)}.
   * @param context the flume {@link Context} used by this source. This can be used
   * by the converter instance to access custom parameters if desired.
   * @param schema the StreamBase {@link Schema} of the stream used by this source.
   * This can be useful to perform schema validation, and to implement field-caching
   * strategies for more efficient conversions. 
   * @throws StreamBaseException
   */
  public void init(Context context, Schema schema) throws StreamBaseException;
        
  /**
   * Converts the StreamBase {@link Tuple} to a Flume {@link Event}.
   * @param tuple the StreamBase tuple to convert.
   * @param event the Flume event into which to place the tuple's contents.
   * @return true if the conversion was successful, false otherwise.
   * @throws StreamBaseException
   */
  public boolean convert(Tuple tuple, Event event) throws StreamBaseException;       
}

If no custom converter is specified in the Flume configuration file, the adapter assumes the payload of the Flume event is a simple byte buffer representing a UTF-8 string which is to be mapped to a tuple field of type String named body or Body.

The adapter also ships with another ready-to-use set of sink and source converters (com.streambase.sb.adapter.flume.ByteArraySinkConverter and com.streambase.sb.adapter.flume.ByteArraySourceConverter, respectively), which treat the Flume event payload as an opaque byte array (mapped to a StreamBase field of type Blob named body or Body).

A sample StreamBase application is included when you install the adapter. The sample demonstrates the use of both the Input and Output Flume adapters. It uses a feed simulation to generate text lines, which are emitted on one port destined to a Flume Output adapter. This adapter converts the text to a Flume event using the default String-based converter and sends it on to the Flume channel. A Flume Input adapter is listening as the Flume sink on the other end of the channel, and converts the received event into a line of text that is placed in a tuple and sent back to the StreamBase application. For further details, refer to the sample's README file.