You can also specify other sources of events. For example, if you want a text file where each line represents a new event, run the following command.

$ flume dump 'text("/etc/services")'

This command reads the file, and then outputs each line as a new event.

	Note
	The default console output escapes special characters with Java-style escape sequences. Characters such as " and \ are prefaced with an extra \.

	Note
	You can try this command with other files such as /var/log/messages, /var/log/syslog, or /var/log/hadoop/hadoop.log also. However, Flume must run with appropriate permissions to read the files.

If you want to tail a file instead of just reading it, specify another source by using tail instead of text.

$ flume dump 'tail("testfile")'

This command pipes data from the file into Flume and then out to the console.

This message appears: "File testfile does not currently exist, waiting for file to appear".

In another terminal, you can create and write data to the file:

$ echo Hello world! >> testfile

New data should appear.

When you delete the file:

$ rm testfile

The tail sink detects this. If you then recreate the file, the tail source detects the new file and follows it:

$ echo Hello world again! >> testfile

You should see your new message appear in the Flume node console.

You can also use the multitail source to follow multiple files by file name:

$ flume dump 'multitail("test1", "test2")'

And send it data coming from the two different files:

$ echo Hello world test1! >> test1
$ echo Hello world test2! >> test2

The tail source by default assumes \n as a delimiter, and excludes the delimiter from events. There are optional line delimiter arguments that allow you to specify arbitrary regular expressions as delimiters and to specify if the delimiter should be part of the prev ious event, next event, or exclude d.

Here are some examples and scenarios to illustrate:

The following example tails a file that requires two or more consecutive new lines to be considered a delimiter. The newlines are excluded from the events.

tail("file", delim="\n\n+", delimMode="exclude")

This example tails a file and uses </a> as a delimiter, and appends the delimiter to the previous event. This could serve as a quick-and-dirty xml record splitter.

tail("file", delim="</a>", delimMode="prev")

Finally, this example tails a file and uses the regex "\n\d\d\d\d" as a delimiter and appends the delimiter to the next event. This could be used to gather lines from a stack dump in a log file that starts with four digits (like a year from a date stamp).

tail("file", delim="\\n\\d\\d\\d\\d", delimMode="next")

Here’s one more example where you use the synth sources to generate events:

$ flume dump 'asciisynth(20,30)'

You should get 20 events, each with 30 random ASCII bytes.

As with files, you can also accept data from well known wire formats such as syslog. For example, you can start a traditional syslog-like UDP server listening on port 5140 (the normal syslog UDP port is the privileged port 514) by running this command:

$ flume dump 'syslogUdp(5140)'

You can feed the source data by using netcat to send syslog formatted data as shown in the example below:

$ echo "hello via syslog"  | nc -u localhost 5140

	Tip
	You may need to press ^C to exit this command.

	Note
	The extra <37> is a syslog wireformat encoding of a message category and priority level.

Similarly, you can set up a syslog-ng compatible source that listens on TCP port 5140 (the normal syslog-ng TCP port is the privileged port 514):

$ flume dump 'syslogTcp(5140)'

And send it data:

$ echo "hello via syslog" | nc -t localhost 5140

	Tip
	You may need to press ^C to exit this command.

Syslog backwards-compatibility allows data normally created from syslog, rsyslog, or syslog-ng to be sent to and processed by Flume.

The Master can be manually started by executing the following command:

$ flume master

After the Master is started, you can access it by pointing a web browser to http://localhost:35871/. This web page displays the status of all Flume nodes that have contacted the Master, and shows each node’s currently assigned configuration. When you start this up without Flume nodes running, the status and configuration tables will be empty.

The web page contains four tables — the Node status table, the Node configuration table, the Physical/Logical Node mapping table, and a Command history table. The information in these tables represent the current global state of the Flume system.

The Master’s Node status table contains the names of all of the Flume Nodes talking to the Master, their current configuration version (initially "none"), their status (such as IDLE), and when it last reported to the Master. The name of each Flume node should be the same as running hostname from Unix prompt.

The Master’s Node configuration table contains the logical name of a node, the configuration version assigned to it, and a specification of its sources and its sinks. Initially, this table is empty, but after you change values you can view this web page to see the updates. There are two sets of columns - - the user entered version/source/sink, and translated version/source/sink. A later section of this guide describes translated configs.

The Master’s Physical/Logical Node mapping table contains the mapping of logical nodes to their physical nodes.

The Master’s Command history table contains the state of commands. In general, commands modify the Master’s global state. Commands are processed serially on a Master and are assigned a unique command ID number. Each command has a state (for example, SUCCEEDED, FAILED, or PENDING), a command line, and a message which often contains information about its execution attempt.

To start a Flume node, invoke the following command in another terminal.

$ flume node_nowatch

	Note
	Normally, you start a node using flume node or run the node as a daemon. For these examples, you disable the watchdog feature by starting with the node_nowatch option. This option enables you to interact with a node via the console. The other options disable stdin.

To check whether a Flume node is up, point your browser to the Flume Node status page at http://localhost:35862/. Each node displays its own data on a single table that includes diagnostics and metrics data about the node, its data flows, and the system metrics about the machine it is running on. If you have multiple instances of the flume node program running on a machine, it will automatically increment the port number and attempt to bind to the next port (35863, 35864, etc) and log the eventually selected port.

If the node is up, you should also refresh the Master’s status page (http:// localhost:35871) to make sure that the node has contacted the Master. You brought up one node (assume the node is named host), so you should have one node listed in the Master’s node status table, and an entry in the logical node mapping table that links the host logical node to the host physical nodes.

When you are using agent/collector roles, you can add the following configuration properties to your flume-site.xml file to set up the default hosts used as collector.

...
<property>
<name>flume.collector.event.host</name>
<value>collector</value>
<description>This is the host name of the default "remote"     collector.
</description>
</property>
  <property>
<name>flume.collector.port</name>
<value>35853</value>
<description>This default tcp port that the collector listens to     in order to receive events it is collecting.
</description>
</property>

...

This will make the agentSink with no arguments default to using flume.collector.event.host and flume.collector.port for their default target and port.

In the following example, a larger setup with several agents push data to a collector. There are seven Flume nodes — six in the agent tier, and one in the collector tier.

An explicit configuration fills in all of the parameters:

agentA : src | agentSink("collector",35853);
agentB : src | agentSink("collector",35853);
agentC : src | agentSink("collector",35853);
agentD : src | agentSink("collector",35853);
agentE : src | agentSink("collector",35853);
agentF : src | agentSink("collector",35853);
collector : collectorSource(35853) | collectorSink("hdfs://namenode/flume/","srcdata");

	Note
	When specifying destinations for agentSinks, use the hostname and port of the target machine. The default name for a node is its hostname. However, if there are multiple logical nodes, you must use the machine’s host name, not the name of the logical node. In the preceding examples, agent[A-F] and collector are the physical host names of the machines where these configurations are running.

You can rely on the default ports set in the configuration files:

agentA : src | agentSink("collector");
agentB : src | agentSink("collector");
agentC : src | agentSink("collector");
agentD : src | agentSink("collector");
agentE : src | agentSink("collector");
agentF : src | agentSink("collector");
collector : collectorSource | collectorSink("hdfs://namenode/flume/","srcdata");

You can rely on the default ports and default collector host:

agentA : src | agentSink
agentB : src | agentSink
agentC : src | agentSink
agentD : src | agentSink
agentE : src | agentSink
agentF : src | agentSink
collector : collectorSource | collectorSink("hdfs://namenode/flume/","srcdata");

	Warning
	Using defaults can make writing data flow configurations more concise, but may obscure the details about how different nodes are connected to each other.

The preceding graph and dataflow spec shows a typical topology for Flume nodes. For reliable delivery, in the event that the collector stops operating or disconnects from the agents, the agents would need to store their events to their respective disks locally. The agents would then periodically attempt to recontact a collector. Because the collector is down, any analysis or processing downstream is blocked.

When you have multiple collectors as in the preceding graph and dataflow spec, downstream progress is still made even in the face of a collector’s failure. If collector B goes down, agent A, agent B, agent E, and agent F continue to deliver events via collector A and collector C respectively. Agent C and agent D may have to queue their logs until collector B (or its replacement) comes back online.

The following configuration partitions the work from the set of agents across many collectors. In this example, each of the collectors specify the same DFS output directory and file prefixes, aggregating all of the logs into the same directory.

agentA : src | agentE2ESink("collectorA",35853);
agentB : src | agentE2ESink("collectorA",35853);
agentC : src | agentE2ESink("collectorB",35853);
agentD : src | agentE2ESink("collectorB",35853);
agentE : src | agentE2ESink("collectorC",35853);
agentF : src | agentE2ESink("collectorC",35853);
collectorA : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorB : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorC : collectorSource(35853) | collectorSink("hdfs://...","src");

When you have multiple collectors writing to the same storage location, instead of having agent C and agent D queue indefinitely, you can instead have them fail over to other collectors. In this scenario, you could have agent C and agent D fail over to collector A and collector C respectively, while periodically checking to see if collector B has returned.

To specify these setups, use agents with failover chains. Similarly to single collector agents, there are three levels of reliability for the failover chain agents: agentE2EChain, agentDFOChain, and agentBEChain.

In the following example, you manually specify the failover chain using agentE2EChain, an agent with end-to-end reliability with multiple failover collectors. agentA in this situation will initially attempt to send to collectorA on port 35853. The second argument in agentA 's sink specifies the collector to fall back onto if the first fails. You can specify an arbitrary number of collectors, but you must specify at least one.

agentA : src | agentE2EChain("collectorA:35853","collectorB:35853");
agentB : src | agentE2EChain("collectorA:35853","collectorC:35853");
agentC : src | agentE2EChain("collectorB:35853","collectorA:35853");
agentD : src | agentE2EChain("collectorB:35853","collectorC:35853");
agentE : src | agentE2EChain("collectorC:35853","collectorA:35853");
agentF : src | agentE2EChain("collectorC:35853","collectorB:35853");
collectorA : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorB : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorC : collectorSource(35853) | collectorSink("hdfs://...","src");

	Note
	In this section, agent[A-F] and collector[A-B] are physical host names.

As in the single collector case, if no port number is specified, the agent defaults to using the flume.collector.port.

agentA : src | agentE2EChain("collectorA","collectorB");
agentB : src | agentE2EChain("collectorA","collectorC");
agentC : src | agentE2EChain("collectorB","collectorA");
agentD : src | agentE2EChain("collectorB","collectorC");
agentE : src | agentE2EChain("collectorC","collectorA");
agentF : src | agentE2EChain("collectorC","collectorB"); collectorA : collectorSource | collectorSink("hdfs://...","src");
collectorB : collectorSource | collectorSink("hdfs://...","src");
collectorC : collectorSource | collectorSink("hdfs://...","src");

	Warning
	The automatic failover chain feature does not currently work when using multiple masters.

Flume also provides a mechanism that automatically assigns failover chains based on how nodes are configured. As collector nodes are assigned in the Flume Master, the Master attempts to distribute the agents evenly amongst the collectors. In the face of failure, each agent is assigned a different failover chain. This mitigates the chances of another collector becoming overloaded in the event of failure of a collector.

To specify a node to use the failover chains, use either the autoE2EChain, autoDFOChain, or autoBEChain agent sink. Because the Master calculates the failover chains, these sinks take no explicit arguments.

agentA : src | autoE2EChain ;
agentB : src | autoE2EChain ;
agentC : src | autoE2EChain ;
agentD : src | autoE2EChain ;
agentE : src | autoE2EChain ;
agentF : src | autoE2EChain ;
collectorA : autoCollectorSource | collectorSink("hdfs://...", "src");
collectorB : autoCollectorSource | collectorSink("hdfs://...", "src");
collectorC : autoCollectorSource | collectorSink("hdfs://...", "src");

The Master updates the configuration of the agents based on the current collectors in the system. When new collectors are added to the system, the Master updates the failover chains of agents to rebalance.

	Note
	If the Master has no nodes with autoCollectorSource as its source, the agent’s automatic chains will report a fail("…") chain which will wait for autoCollectorSource s to be specified. If the nodes are not mapped, they will report a different fail sink notifying you that the node is unmapped (isn’t associated with a host/port).

	Tip
	You can see the translation of the auto*Chain configuration in the node configuration table under the translated configuration column. This is a little declarative specification of the failure recovery behavior of the sink. More details on this are in the Advanced section of this guide, and in future revisions the translations for the agents and other chains will also be presented.

The logical node abstraction allows for each JVM instance (a physical node) to contain multiple logical nodes. This allows for the processing of many source sink combinations on many threads of execution to occur on a single JVM instance.

Each logical node has a name that may be completely different from its physical name or hostname. You now need new operations that enable you to spawn a new node, map logical nodes to physical nodes, and decommission existing logical nodes.

	Note
	The following commands are entered via the web interface using the "raw command" web page on the Master. You might prefer using the Flume command shell (described in a later section) for these operations. The same commands described in this section can be entered in web interface or entered at the command shell by prefixing the command with exec.

Suppose that initially you know you want an agent-collector topology, but you don’t know the particular names of the exact machines. For now, you can specify the configuration of the logical nodes without specifying any physical machine names.

agent1 : _source_ | autoBEChain ;
collector1 : autoCollectorSource | collectorSink("hdfs://....") ;

Later you learn that host1 is the name of the agent1 machine and host2 is the name of the collector machine. You can map logical nodes onto the physical Flume instances on host1 and host2 by issuing the following map commands:

map host1 agent1
map host2 collector1

Afterwards, the node status table should display a new row of information for each logical node. Each logical node reports its own execution state, configuration, and heartbeat. There is also a new entry in the logical node mapping table showing that the logical node has been placed on the specified physical node. To configure the node’s sources and sinks, use exactly the same mechanisms described in the previous sections.

You can also remove a logical node by using the decommission command. Suppose you no longer needed agent1 and wanted to "turn it off". You can do so by entering the following command:

decommission agent1

This terminates the thread and removes the configuration associated with a logical node, and the mapping between the logical node and physical node.

You can also move a logical node from one physical node to another by first unmapping a logical node and then mapping it on another physical node. In this scenario, you change the collector1 from being on host2 to host3.

unmap host2 collector1

At this point, the logical node mapping is removed, and collector1 is not active anywhere. You can then map collector1 onto host3 by using the map command:

map host3 collector1

	Note
	Logical nodes are not templates — if you want to have the same source/sink pairs on a particular physical node, you need to have a logical node for each. When adminstering many logical nodes it is often useful to write a script that generates configurations and unique individual logical node names. Using the part of a host name is a common pattern.

	Warning
	The logical sources and logical sinks feature does not currently work when using multiple masters.

In the previous example, we used two abstractions under-the-covers that allow the specifications of a graph topology for communications without having to use physical hostnames and ports. These abstractions — the logical source and logical sink — allow you to create a different graph topology without having to know physical machines until they are mapped.

Suppose you have two nodes producing data and sending it to the consumer:

dataProducer1 : console | logicalSink("dataConsumer") ;
dataProducer2 : console | logicalSink("dataConsumer") ;
dataConsumer : logicalSource | console ;

Note that in this example, the destination argument is the logical name of the node and not a specific host/port combination.

To implement these features, there is a generalized mechanism where users enter logical configurations that are translated by the Master to a physical configuration.

When the logical nodes get mapped to physical nodes:

map host1 dataProducer1
map host2 dataProducer2
map host3 dataConsumer

and after the Master learns the host names (the host1, host2, and host3 machine’s heartbeat against the Master), the Master has enough information to translate configurations with physical hostnames and ports. A possible translation would replace the logicalSource with a rpcSources and the logicalSink with an rpcSinks:

dataProducer1 : console | rpcSink("host3",56789) ;
dataProducer2 : console | rpcSink("host3",56789) ;
dataConsumer : rpcSource(56789) | console ;

In fact, auto agents and collectors, are another example of translated sources and sinks. These translate auto*Chain sinks and collectorSource into a configuration that uses logicalSinks and logicalSources which in turn are translated into physical rpcSource and rpcSinks instances.

	Tip
	Translations are powerful and can be fairly smart; if new collectors are added, they will become new failover options. If collectors are removed, then the removed collectors will be automatically replaced by other failover nodes.

	Warning
	The automatic flow isolation feature does not currently work when using multiple masters.

What happens if you want collect different kinds of data from the same physical node? For example, suppose you wanted to collect httpd logs as well as syslog logs from the same physical machine. Suppose also you want to write all of the syslog data from the cluster in one directory tree, and all of the httpd logs from the cluster in another.

One approach is to tag all the data with source information and then push all the data down the same pipe. This could then be followed by some post- processing that demultiplexes (demuxes) the data into different buckets. Another approach is to keep the two sets of data logically isolated from each other the entire time and avoid post processing.

Flume can do both approaches but enables the latter lower-latency approach, by introducing the concept of grouping nodes into flows. Since there are logical nodes that allow for multiple nodes on a single JVM, you can have a node for each different kinds of produced data.

The following example shows how flows can be used. Start by having six logical nodes in the system.

fooNode1 : fooSrc | autoBEChain ;
barNode1 : barSrc | autoBEChain ;
forNode2 : fooSrc | autoBEChain ;
barNode2 : barSrc | autoBEChain ;
fooConsumer : autoCollectorSource | collectorSink("hdfs://nn/foodir") ;
barConsumer : autoCollectorSource | collectorSink("hdfs://nn/bardir") ;

In this scenario, there are two physical machines that produce both kinds of data — foo data and bar data. You want to send data to single collector that collects both foo data and bar data and writes it to different HDFS directories. You could then map the nodes onto physical nodes:

map host1 fooNode1
map host1 barNode1
map host2 fooNode2
map host2 barNode2
map host3 fooConsumer
map host3 barConsumer

Figure 1. Flume Flows: Single Flow

This setup essentially instantiates the first approach. It mixes foo and bar data together since the translation of autoBEChain would see two collectorSources that the Master considers to be equivalent. Foo data will likely be sent to the barConsumer and bar data will likely be sent to fooConsumer.

You really wanted to separate sources of information into logically isolated streams of data. Flume provides a grouping abstraction called a flow. A flow groups particular logical nodes together so that the different logical data types remain isolated.

More concretely, it allows for a different failover chain for each kind of data in the Flume cluster. The auto*Chain based agents would only send data to collectors in the same flow group. This isolates data so that it only flows to nodes within the group.

Currently, the compact form of the configuration language does not allow you to specify flows. Instead you must add an extra argument to the config command to specify a flow.

This example shows commands that would be entered in the Flume shell without flow group information. In this case all of the nodes are in the same flow.

exec config fooNode1 fooSrc autoBEChain
exec config barNode1 barSrc autoBEChain
exec config fooNode2 fooSrc autoBEChain
exec config barNode2 barSrc autoBEChain
exec config fooConsumer autoCollectorSource 'collectorSink("hdfs://nn/foodir")'
exec config barConsumer autoCollectorSource 'collectorSink("hdfs://nn/bardir")'

Now using the following commands you can specify flows by adding an extra parameter after the node name. In this example we have two flows: flowfoo and flowbar. flowfoo contains fooNode1, fooNode2 and fooConsumer. flowbar contains barNode1, barNode2 and barConsumer.

exec config fooNode1 flowfoo fooSrc autoBEChain
exec config barNode1 flowbar barSrc autoBEChain
exec config fooNode2 flowfoo fooSrc autoBEChain
exec config barNode2 flowbar barSrc autoBEChain
exec config fooConsumer flowfoo autoCollectorSource 'collectorSink("hdfs://nn/foodir")'
exec config barConsumer flowbar autoCollectorSource 'collectorSink("hdfs://nn/bardir")'

Figure 2. Flume Flows: Multiple Flows

By using these commands, the data from fooNode1 and fooNode2 will only be sent to fooConsumer, and barNode1 and barNode2’s data will only be sent to barConsumer. Data from one node is not mixed with other data from other nodes unless explicitly connected.

	Tip
	In practice it is a good idea to use different node names and different flow ids for different kinds of data. When node names are reused, the default behavior is to attempt to recover from failures assuming that leftover data from a crashed execution or previous source/sink configuration version are still producing the same kind of data.

This section introduced logical nodes, logical sources, logical sinks, and flows and showed how these abstractions enable you to automatically deal with manageability problems.

The translation mechanism can be quite powerful. When coupled with metrics information, this could be used to perform automated dynamic configuration changes. A possible example would be to automatically commission or decommission new collectors to match diurnal traffic and load patterns.

The Flume Master can be run in one of two ways.

Large production deployments of Flume should run a distributed Master so that inevitable machine failures do not impact the availability of Flume itself. For small deployments the issue is less clear-cut - a distributed Master means reserving more computing resources that could be used instead for nodes or other services, and it is possible to recover from many failure modes in a timely manner with human intervention. The choice between distributed and standalone Masters is ultimately dependent both on your use case and your operating requirements.

Whether the Flume Master starts in distributed or standalone mode is indirectly controlled by how many machines are configured to run as Master servers. To run in standalone mode, a single configuration property flume.master.servers must be set:

<property>
<name>flume.master.servers</name>
<value>hostA</value>
</property>

The value of flume.master.servers is a comma-separated list of all the machine names (or IP addresses) that will be Master servers. If this list contains only machine name, the Flume Master will start in standalone mode. If there is more than one machine name in the list, the Flume Master will start in distributed mode.

There’s no other configuration required for standalone mode. Flume will use reasonable default values for any other master-related variables. To start the Master, from the command prompt type:

$ flume master

from $FLUME_HOME. A number of log messages should print to the screen. After the server is running, you can check that everything is working properly by visiting the web interface at http://master-node-ip:35871/, where master-node-ip is the IP address (or hostname) of the Master node. If you see a web page, the Master is running.

<property>
<name>flume.master.servers</name>
<value>masterA,masterB,masterC</value>
</property>

How many machines do I need? The distributed Flume Master will continue to work correctly as long as more than half the physical machines running it are still working and haven’t crashed. Therefore if you want to survive one fault, you need three machines (because 3-1 = 2 > 3/2). For every extra fault you want to tolerate, add another two machines, so for two faults you need five machines. Note that having an even number of machines doesn’t make the Flume Master any more fault-tolerant - four machines only tolerate one failure, because if two were to fail only two would be left functioning, which is not more than half of four. Common deployments should be well served by three or five machines.

Note. flume.master.serverid is the only Flume Master property that must be different on every machine in the ensemble. *

masterA. 

<property>
<name>flume.master.serverid</name>
<value>0</value>
</property>

masterB. 

<property>
<name>flume.master.serverid</name>
<value>1</value>
</property>

masterC. 

<property>
<name>flume.master.serverid</name>
<value>2</value>
</property>

The value for flume.master.serverid for each node is the index of that node’s hostname in the list in flume.master.ensemble, starting at 0. For example masterB has index 1 in that list. The purpose of this property is to allow each node to uniquely identify itself to the other nodes in the Flume Master.

This is all the configuration required to start a three-node distributed Flume Master. To test this out, we can start the Master process on all three machines:

[flume@masterA] flume master

[flume@masterB] flume master

[flume@masterC] flume master

Each Master process will initially try and contact all other nodes in the ensemble. Until more than half (in this case, two) nodes are alive and contactable, the configuration store will be unable to start, and the Flume Master will not be able to read or write configuration data.

You can check the current state of the ensemble by inspecting the web page for any of the Flume Master machines which by default will be found at, for example, http://masterA:35871.

The Flume Master stores all its data in a configuration store. Flume has a pluggable configuration store architecture, and supports two implementations.

Flume and Apache ZooKeeper . Flume relies on the Apache ZooKeeper coordination platform to provide reliable, consistent, and persistent storage for node configuration data. A ZooKeeper ensemble is made up of two or more nodes which communicate regularly with each other to make sure each is up to date. Flume embeds a ZooKeeper server inside the Master process, so starting and maintaining the service is taken care of. However, if you have an existing ZooKeeper service running, Flume supports using that external cluster as well.

In almost all cases, you should use the ZBCS. It is more reliable and fault-tolerant, and will recover configurations after a restart. It is compatible with both standalone and distributed deployments of the Flume Master.

The MBCS is appropriate if you are experimenting with Flume and can stand to lose configuration if the machine fails.

ZBCS is the default configuration store. The choice of which configuration store to use is controlled by the flume.master.store system property.

<property>
<name>flume.master.store</name>
<value>zookeeper</value>
</property>

If set to memory, the Flume Master will use MBCS instead. This is only supported in standalone mode.

Most deployments using the ZBCS can use Flume’s default configuration. However, where more control over the precise configuration of the Flume Master is needed, there are several properties that you can set.

Flume Master servers also use a gossip protocol to exchange information between themselves. Each server periodically wakes and picks another machine to send new data to. This protocol by default uses TCP port 57890, but this is controlled via the flume.master.gossipport property:

<property>
<name>flume.master.gossip.port</name>
<value>57890</value>
</property>

In standalone mode, there is no need to use gossip, so this port is unused.

One property needs to be set to configure a Flume Node to connect to multiple Masters: flume.master.servers.

<property>
<name>flume.master.servers</name>
<value>masterA,masterB,masterC</value>
</property>

The nodes connect over the port flume.master.heartbeat.port on each machine in the Flume Master - this is the port that the Master servers listen on for node heartbeats.

If a Master server fails, nodes will automatically fail over to the next randomly selected Master server that they can establish a connection to.

To be added

To be added

You can set the default output format by etting the flume.collector.output.format property set in the flume-site.xml file. The output formats are:

Here is an example of a property added to the flume-site.xml file:

<property>
  <name>flume.collector.output.format</name>
  <value>avrojson</value>
  <description>This is the output format for the data written to the
  collector.  There are several formats available:
    avro - Avro Native file format.  Default currently is uncompressed.
    avrodata - this outputs data as an avro binary encoded data
    avrojson - this outputs data as json encoded by avro
    debug - this is a format for debugging
    json - this outputs data as json
    log4j - outputs events in a pattern similar to Hadoop's log4j pattern
    raw - Event body only.  This is most similar to copying a file but
      does not preserve any uniqifying metadata like host/timestamp/nanos.
    seqfile - this is the hadoop sequence file format with
       WritableEventKeys and WritableEvent objects.
    syslog - outputs events in a syslog-like format
  </description>
</property>

Some sinks have an optional output format argument. These include the console, text, customdfs/formatDfs, escapedCustomDfs/escapedFormatDfs, and collectorSink.

In these cases, we will be focusing on the optional format argument.

collectorSink( "dfsdir","prefix"[, rollmillis[, format]])
text("file"[,format])
formatDfs("hdfs://nn/file" [, format])
escapedFormatDfs("hdfs://nn/file" [, format])

This mechanism is more flexible because it allows for output formats with arguments. Currently, the seqfile output format supports arugments for specifying an sequence file internal compression codec. Codecs available for use are the same as those available to hadoop — generally these include gzip, bzip2, and can include others that have been plugged in such lzo or snappy.

The compression codec is specified by adding an argument to output format. For example, the seqfile output format accepts a compression codec argument. So, to specify an output format that is a sequence file using the bzip2 codec, the output format would be seqfile("bzip2"). To write bzip2 compressed sequence files, using a formatDfs sink, we would specify formatDfs("hdfs://nn/dir/file", seqfile("bzip2"))

	Note
	The old syntax requires specifying a Flume String (ex: console("avrojson"), console("seqfile") has been deprecated but will be supported for a few more versions. The preferred arugment is now a "function". (ex: console(avrojson), console(seqfile("bzip2"))).

You can start the FlumeShell by running flume shell in a terminal window. The connect command can be used to establish a connection to any Master server.

hostname:~/flume$ flume shell
[flume (disconnected)] connect localhost:35873
Connecting to Flume master localhost:35873...
[flume localhost:35873]

hostname:~/flume$ flume shell -c localhost:35873
Connecting to Flume master localhost:35873...
[flume localhost:35873]

The command line parameters for the Flume Shell are as follows:

usage: FlumeShell [-c <arg>] [-e <arg>] [-q] [-s <arg>]
 -?         Command line usage help
 -c <arg>   Connect to master:port
 -e <arg>   Run a single command
 -q         Run in quiet mode - only print command results
 -s <arg>   Run a FlumeShell script

The FlumeShell makes scripting Flume possible - either by using a single invocation with -e or by running a script of commands with -s. It is also possible to pipe stdin to the FlumeShell as in the following example:

echo "connect localhost:35873\ngetconfigs\nquit" | flume shell -q

Flume Commands. You can press Tab any time for some hints on available commands. If you start typing a command you can use TAB to complete command.

Exec and Submit commands. Both the web form and the FlumeShell are interfaces to the same command processing infrastructure inside Flume. This section introduces the FlumeShell and show how you can use it to make administering Flume more simple.

These commands are issued and run as if run from the master. In the command shell they have the form:

exec command [arg1 [_arg2 [ … ] ] ]

submit command [arg1 [_arg2 [ … ] ] ]

Complex arguments like those with spaces, or non alpha-numeric characters can be expressed by using "double quotes"s and `single quotes’s. If enclosed in double quotes, the bodies of the strings are Java string unescaped. If they are enclosed in single quotes, arbitrary characters can be included except for the ' character.

exec commands block until they are completed. submit commands are asynchronously sent to the master in order to be executed. wait are essentially joins for recently submit ted commands.

Three special sinks are FanOutSinks, FailoverSinks, and RollSinks.

Fanout sinks send any incoming events to all of the sinks specified to be its children. These can be used for data replication or for processing data off of the main reliable data flow path. Fanout is similar to the Unix tee command or logically acts like an AND operator where the event is sent to each subsink.

The syntax for a FanoutSink is :

[ console, collectorSink ]

FailoverSinks are used to handle failures when appending new events. FailoverSinks can be used to specify alternate collectors to contact in the event the primary collector fails, or a local disk sink to store data until the primary collector recovers. Failover is similar to exception handling and logically acts like an OR operator. If a failover is successful, one of the subsinks has received the event.

The syntax for a FailoverSink is :

< logicalSink("collector1") ? logicalSink("collector2") >

So, you could configure node "agent1" to have a failover to collector2 if collector1 fails (for example, if the connection to collector1 goes down or if collector1's HDFS becomes full):

agent1 : source | < logicalSink("collector1") ? logicalSink("collector2")
> ;

Roll sink opens and closes a new instance of its subsink every millis milliseconds. A roll is an atomic transition that closes the current instance of the sub-sink, and then opens a new instance of as an escape sequence to differentiate data in different roll periods. This can be used to make every roll produce a new file with a unique name.

The syntax for a roll sink is:

roll(millis) sink

These can be composed to have even richer behavior. For example, this sink outputs to the console and has a failover collector node.

[ console, < logicalSink("collector1") ? logicalSink("collector2") > ]

This one rolls the collector every 1000 milliseconds writing to a different HDFS file after each roll.

roll(1000) [ console, escapedCustomDfs("hdfs://namenode/flume/file-%{rolltag}") ]

Fan out and failover affect where messages go in the system but do not modify the messages themselves. To augment or filter events as they pass through the dataflow, you can use sink decorators.

Sink decorators can add properties to the sink and can modify the data streams that pass through them. For example, you can use them to increase reliability via write ahead logging, increase network throughput via batching/compression, sampling, benchmarking, and even lightweight analytics.

The following simple sampling example uses an intervalSampler which is configured to send every 10th element from source "source" to the sink "sink":

flumenode: source | intervalSampler(10) sink;

	Note
	There is an older more verbose syntax for specifiying decorators that wraps decorators and sinks with { deco ⇒ sink } syntax. The old syntax remains valid.

Here’s an example that batches every 100 events together.

flumenode: source | batch(100) sink;

Like fanout and failover, decorators are also composable. Here is an example that creates batches of 100 events and then compresses them before moving them off to the sink:

flumenode: source | batch(100) gzip sink;

As of v0.9.3, a new more generic collector sink that better encapsulates the "collectorMagic" functionality. This allows you to specify arbitrary sinks (or mulitple sinks) using the flume specification language. So the equivalent of

collectorSink("xxx","yyy-%{rolltag}", 15000)

is now:

collector(15000) { escapedCustomDfs("xxx","yyy-%{rolltag}") }

Here is an example of newly expressable functionality:

collector(15000) { [ escapedCustomDfs("xxx","yyy-%{rolltag}"), hbase("aaa", "bbb-%{rolltag}"), elasticSearch("eeee","ffff") ] }

This would send data to all three destinations, and only send an ack if all three were successful. Combined with some recovery decorators (stubborn*, insistent*), this can be used to precisely express potentially complex failure recovery policies.

Internally, Flume translates sinks into compositions of other simpler decorators, failovers, rollers to add properties. The proper compositions create pipelines that provide different levels of reliability in Flume.

	Note
	This section describes how agents work, but this version of Flume does not expose the translations the same way the auto*Chains are exposed. A future version of Flume will expose these details. The exact translations are still under development.

Suppose you have nodes using the different agent sinks:

node1 : tail("foo") | agentE2ESink("bar");
node2 : tail("foo") | agentDFOSink("bar");
node3 : tail("foo") | agentBESink("bar");

In the translation phases, agentE2ESink is actually converted into these Flume sinks:

node1 : tail("foo") | ackedWriteAhead lazyOpen stubbornAppend logicalSink("bar") ;
node2 : tail("foo") | < lazyOpen stubbornAppend logicalSink("bar") ?  diskFailover insistentOpen lazyOpen stubbornAppend logicalSink("bar") >;
node3 : tail("foo") | lazyOpen  stubbornAppend  logicalSink("bar");

ackedWriteAhead is actually a complicated decorator that internally uses rolls and some other special decorators. This decorator interface allows us to manually specify wire batching and compression options. For example, you could compress every 100 messages using gzip compression:

node1 : tail("foo") | ackedWriteAhead batch(100) gzip lazyOpen stubbornAppend logicalSink("bar");

collectorSink("xxx","yyy",15000) is also a bit complicated with some custom decorators to handle acks. Under the covers however, it depends on a roller with a escapedCustomDfsSink inside of it.

roll(15000) { collectorMagic escapedCustomDfs("xxx", "yyy-%{rolltag}") }

Another place translations happen is with logical nodes. Lets start of with a few nodes:

node1 : tail("foo") | { .... => logicalSink("node2") };
node2 : logicalSource | collectorSink("...");

The translation mechanisms converts logicalSources and logicalSinks into lower level physical rpcSources and rpcSinks. Lets assume that node1 is on machine host1 and node2 is on machine host2. After the translation you end up with the following configurations:

node1 : tail("foo") | { .... => rpcSink("host2",12345) };
node2 : rpcSource(12345) | collectorSink("..."");

Suppose that you swap the mapping so that node2 is now on host1 and node1 is on host2.

# flume shell commands
exec unmapAll
exec map host1 node2
exec map host2 node1

The original configuration will now be translated to:

node1 : tail("foo") | { .... => rpcSink("host1",12345) };
node2 : rpcSource(12345) | collectorSink("..."");

Extraction operations are also available to extract values from logs that have known structure.

One example is the regex decorator. This takes three arguments: a regular expression, an index, and an attribute name which allows the user to use a extract a particular regex group out of the body of the event and write it as the value of the specified attribute.

Similarly the split decorator also takes three arguments: a regular expression, an index, and an attribute name. This splits the body based on the regular expression, extracts the text group after the instance of the separator, and writes the value to the specified attribute. One can think of this as a much simplified version of the awk Unix utility.

Flume enforces an invariant that prevents the modification of an attribute that has already been written upstream. This simplifies debugging of dataflows, and improves the visibility of data when debugging.

If many stages are used, however, frequently a lot of extra metadata ends up being moved around. To deal with this, a few extra operations are available.

A select operation is available. This operation is like SQL select, which provides a relational calculus’s set projection operation that modifies an event so that the specified metadata fields are forwarded.

A mask operation is also available that forwards all metadata attributes except for the attributes specified.

A format decorator is also available that uses the escape mechanisms to rewrite the body of an event to a user customizable message. This is useful for outputting summary data to low volume sources. Example: writing summary information out to an IRC channel periodically.

To simplify things for users, you can assign a particular role to a logical node — think of these as an "automatic" specification that have default settings. Two roles currently provided are the agent role and the collector role. Since there are many nodes with the same role, each of these are called a tier. So, for example, the agent tier consists of all the nodes that are in the agent role. Nodes that have the collector role are in the collector tier.

Agents and collectors roles have defaults specified in the conf/flume-site.xml file. Look at the conf/flume-conf.xml file for properties prefixed with flume.agent.\* and flume.collector.\* for descriptions of the configuration options.

Each node maintains it own node state and has its own configuration. If the master does not have a data flow configuration for the logical node, the logical node will remain in IDLE state. If a configuration is present, the logical node will attempt to instantiate the data flow and have it work concurrently with other data flows.

This means that each machine essentially only lives in one tier. In a more complicated setup, it is possible to have a machine that contains many logical nodes, and because each of these nodes can take on different roles, the machine lives in multiple tiers.

With thriftSink and thriftSource, data can be sent through multiple nodes. If ack injection and ack checking decorators are properly inserted, you can achieve reliability.

Flume uses threading to support reconfiguration and multiple logical nodes. Sources, sinks, and decorator are extensions that can use queues for buffering and can use clock sleeps for periodic rolls and retries. Because we use these blocking operations and because the internal concurrency control mechanisms we use can cause deadlocks or hangs, there are several rules that need to be followed and enforced by test cases in order to be a well-behaved source, sink, or decorator.

	Note
	This section is a draft and not exhaustive or completely formal. More restrictions may be added to this in the future.

Semantics of sources. Sources must implement 4 methods

Along with these signatures, each of these methods can also throw RuntimeExceptions. These exceptions indicate a failure condition and by default will make a logical node shutdown in ERROR state. These error messages are user visible and it is important that they have actionable descriptions about the failure conditions without having to dump stack. Thus, NullPointerExceptions are not acceptable — they are not descriptive or helpful without their stack traces.

Some examples of valid runtime exceptions include invalid arguments that prevent a source from opening and invalid state infractions (attempting to next on a closed source).

8.4.1.1. Simple source semantics.

Simple sources are assumed to be sources whose open and close operation happen quickly and do not block for long periods of time. The longest pauses tolerated here on the order of 10s (default time for a failing DNS lookup).

The constructor for the sources should be callable without grabbing resources that can block or that require IO such as network connectors or file handles. If there are errors due to configuration settings that can be caught in the constructor, an IllegalArgumentException should be thrown.

open() is a call that grabs resources for the source so that the next() call can be made. The open call of a simple source should attempt to fail fast. It can throw a IOException or a RuntimeException such as IllegalStateException. Open should only be called on a CLOSED sink — if a sink is opened twice, the second call should throw an IOException or IllegalStateException.

next() is a call that gets an event from an open source. If a source is not open, this should throw an IllegalStateException. This call can and will often block. If next() is blocking, a call to close() should unblock the next() by having it exit cleanly returning null. Many resources such as TCP-sockets (and ideally rpc frameworks) default to throwing an exception on blocked network reads (like a next() call) when close()'d.

close() is a call that releases resources that the open call of a source grabs. The close() call itself blocks until all resources are released. This allows a subsequent open() in the same thread to not fail due to resource contention (ex, closing a server socket on port 12345 should not return until port 12345 is ready to be bound again).

getReport() returns a ReportEvent. These values should be available regardless if the node is open or closed and this call should not block the other source by other calls (due to potential lock inversion issues). The values retrieved are ideally atomically grabbed but this is not required as long as no errors are caused by racy execution.

If a source is opened or closed multiple times, it is up to the source to determine if values are reset or persisted between open/close cycles.

8.4.1.2. Buffered source semantics

Some sources have in memory queues or buffers stored persistently on disk. The guiding principle here is that on close(), buffered sources should prevent new data from entering and attempt to flush the buffered data. Also any subordinate threads should be released before close returns. If no progress is being made on close(), for a given period of time (30s is the default currently) the controlling thread will for an Thread.interrupt() call. The source should be able to handle InterruptedExceptions and percolate interrupted status up the returning call stack.

	Note
	In v0.9.1 and v0.9.2, interruptions when caught should be handled by re-flagging the Thread’s interrupted flag (a call to Thread.currentThread().interrupt()) and then throwing an IOException. The API for extensions will likely change in the future to throw either an IOException or an InterruptedException.

open() no change.

close() This call is usually made from a separate thread than the open() or next() calls. Since this call should block until resources are freed, it should attempt to flush its buffers before returning. For example, so if a network source has some buffered data, the network connection should be closed to prevent new data from entering, and then the buffered data should be flushed. The source should be able to handle InterruptedExceptions and percolate interrupted status up the returning call stack and indicate an error state.

A next() call while in CLOSING state should continue pulling values out of the buffer until it is empty. This is especially true if the next() call is happening in the main driver thread or a subordinate driver thread. One mechanism to get this effect is to add a special DONE event to the queue/buffer that indicates a clean exit.

getReport() ideally includes metric information like the size of the queues in the source, and the number of elements in the queue.

Semantics of sinks and decorators. Sinks and decorators must implement 4 methods

• void open() throws IOException
• void append(Event e) throws IOException
• void close() throws IOException
• ReportEvent getReport()

Along with these signatures, each of these methods can also throw RuntimeExceptions. Run-time exceptions indicate a failure condition and by default will make a logical node shutdown in ERROR state. These error messages are user visible and it is important that they have helpful descriptions about the failure conditions without having to dump stack. Thus, NullPointerExceptions are not acceptable — they are not descriptive or helpful without their stack traces.

8.4.1.3. Simple Sinks.

Simple sinks are assumed to have open, close, and append operations that happen quickly and do not block for long periods of time. The longest pauses tolerated here on the order of 10s (default time for a failing DNS lookup).

The constructor for the sinks and decorators should be callable without grabbing resources that can block or that require IO such as network connectors or file handles. If there are errors due to configuration settings that can be caught in the constructor, an IllegalArgumentException should be thrown.

open() is a call that grabs resources for the sink or decorator so that the append(Event) call can be made. If there are errors due to configuration settings not detectable without IO in the constructor, open() should attempt to fail fast and throw a IOException or RuntimeException such as IllegalStateException or IllegalArgumentException. Open should only be called on a CLOSED sink — if a sink is opened twice, the second call should throw an IOException or IllegalStateException.

append() is a call that delivers a event. If a sink is not open, this should throw an IllegalStateException.

If a normal decorator fails to open or to append because of an internal failure or a subsink fails to open, the decorator should release its resources attempt to close the subsink and then throw an exception. There are some sink/decos that specifically manipulate these semantics — this needs to be done with care.

close() is a call that releases resources that the open call of a source grabs. If open() or next() is blocking, a call to this should unblock the call and have them exit. close() should be called called on an open sink, but we allow a closed sink to have close() called on it without throwing an exception (generally LOG warning however).

getReport() returns a ReportEvent. These values should be available regardless if the node is open or closed and this call should not cause get blocked by other calls (due to potential lock inversion issues). The values retrieved are ideally atomically grabbed but this is not required as long as no errors are caused by racy execution. If a sink is opened or closed multiple times, it is up to the sink to determine if values are reset or persisted between open/close cycles.

8.4.1.4. Buffered sink and decorator semantics

Some sinks have queues or buffers stored in memory or persistently on disk. The guiding principle here is that buffered sinks should attempt to flush its buffers when prompted to close(). This needs to be balanced with the requirement that sinks and decorated sinks should attempt to close in a relatively quick fashion.

open() Since this sink isn’t open this generally means there is no buffered data. However, an open() on a sink or decorator with persistent data should attempt to recover data and enqueue it in the open() call. Examples of these include a DFO or WAL log recovering dfo/wla logs, when a network subsink is down.

An append() call may buffer data before sending it (such as a batching decorator). A close() call, should attempt to append buffered data to its (sub)sink before executing the close. Also, any subordinate threads should be stopped before shutdown.

If no progress is being made on close(), for a given period of time (30s is the default currently) it will be interrupted and should handle abruptly exiting because of this interruption.

8.4.1.5. Retries, sleeps, and unclean exits.

Retry on open semantics

Retry on Append semantics

Some decorators introduce retries and sleeps. An author who uses these needs to ensure that these operations behave well on open, append, and close. When combined with buffering sinks, flushing a buffer on close may not be possible! (ex: wal trying to send data to a dead network connection). This means these sinks/decos need a way to exit abruptly and report an error. There are two operations that make these cases need to be handled: unbounded retries and unbounded/long sleeps/await/blocking.

Some decorators have potentially unbounded retry semantics. For example, InsistentOpen, InsistentAppend, and FailoverSinks have the potential to retry open() and append() calls an unbounded number of times. These decorators can also be wrapped with in others — this means we need to be able to percolate the hard exit and bypass the retry logic.

To do this we require that any sink/deco that has retry logic must check for hard exit before retrying. These sinks must to propagate the hard exit interruption to its upstream decorators (in case they have retry logic!).

Some sinks have the potential to backoff or sleep for long or potentially unbounded amounts of time. Code using sleeps or synchronization operations such as waiting on latches (CountDownLatch.await()) or thread sleeps (Thread.sleep()) must properly handle interruptions. Since these yielding operations are usually only used in retry operations (which meant there was a failure), the sink/deco needs to propagate the interruption and fail with error.

There are some ramifications of these semantics. Care must be taken with locking open, close, and append operations. If there are any sleeps or blocking open() operations (ex: InsistentOpen, FailoverSinks), ideally a close call will cause it to shutdown, it and the open call should get unblocked. append()

The sink signaled to be closed but blocked on append() or open() should exit in a reasonable amount of time — ideally within a few heartbeats (5s is the default, so ideally <30s). If the sink exits and its buffers are empty, it should do a normal successful return. If there were unflushed events, it should return error by throwing an exception. If there are subordinate threads, these should be terminated before close returns.

First you need to have permissions to use the kadmin program and the ability to add to principals to the KDCs.

$ kadmin -p <adminuser> -w <password>

If you entered this correctly, it will drop you do the kadmin prompt

kadmin:

Here you can add a Flume principal to the KDC

kadmin: addprinc flume
WARNING: no policy specified for flume@REALM.COM; defaulting to no policy
Enter password for principal "flume@REALM.COM":
Re-enter password for principal "flume@REALM.COM":
Principal "flume@REALM.COM" created.
kadmin:

You also need to add principals with hosts for each Flume node that will directly write to HDFS. Since you will be exporting the key to a keytab file, you can use the -randkey option to generate a random key.

kadmin: addprinc -randkey flume/host.com
WARNING: no policy specified for flume/host.com@REALM.COM; defaulting to no policy
Principal "flume/host.com@REALM.COM" created.
kadmin:

	Note
	Hadoop’s Kerberos implementation requires a three part principal name — user/host@REALM.COM. As a user you usually only need the user name, user@REALM.COM.

You can verify that the user has been added by using the kinit program, and entering the password you selected. Next you can verify that you have your Ticket Granting Ticket (TGT) loaded.

$ kinit flume/host.com
Password for flume/host.com@REALM.COM:
$ klist
Ticket cache: FILE:/tmp/krb5cc_1016
Default principal: flume/host.com@REALM

Valid starting     Expires            Service principal
09/02/10 18:59:38  09/03/10 18:59:38  krbtgt/REALM.COM@REALM.COM


Kerberos 4 ticket cache: /tmp/tkt1016
klist: You have no tickets cached
$

You can ignore the Kerberos 4 info. To "logout" you can use the kdestroy command, and then verify that credentials are gone by running klist.

$ kdestroy
$ klist
klist: No credentials cache found (ticket cache FILE:/tmp/krb5cc_1016)


Kerberos 4 ticket cache: /tmp/tkt1016
klist: You have no tickets cached
$

Next to enable automatic logins, we can create a keytab file so that does not require manually entering a password.

	Warning
	This keytab file contains secret credentials that should be protected so that only the proper user can read the file. After created, it should be in 0400 mode (-r--------) and owned by the user running the Flume process.

Then you can generate a keytab file (int this example called flume.keytab) and add a user flume/host.com to it.

kadmin: ktadd -k flume.keytab flume/host.com

	Note
	This will invalidate the ability for flume/host.com to manually login of the account. You could however have a Flume user does not use a keytab and that could log in.

	Warning
	ktadd can add keytab entries for mulitple principals into a single file and allow for a single keytab file with many keys. This however weakens the security stance and may make revoking credentials from misbehaving machines difficult. Please consult with your security administrator when assessing this risk.

You can verify the names and the version (KVNO) of the keys by running the following command.

$ klist -Kk flume.keytab
Keytab name: FILE:flume.keytab
KVNO Principal
---- --------------------------------------------------------------------------
   5 flume/host.com@REALM.COM (0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa)
   5 flume/host.com@REALM.COM (0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb)
   5 flume/host.com@REALM.COM (0xcccccccccccccccc)
   5 flume/host.com@REALM.COM (0xdddddddddddddddd)

You should see a few entries and your corresponding keys in hex after your principal names.

Finally, you can use kinit with the flume@REALM.COM principal to interactively do a Kerberos login and use the Hadoop commands to browse HDFS.

$ kinit flume
Password for flume@REALM.COM:  <-- enter password
$ hadoop dfs -ls /user/flume/

TCP ports are used in all situations.

When using auto chains for collectors, each collector source will automatically search for a free port by incrementing starting from port flume.collector.port and incrementing the port number.

When there is more than one physical node on a machine, the node status web server will attempt to bind to port flume.node.http.port. If the bind fails, the port number will be incremented and the bind retried until a free port is found. The port that is eventually bound to is logged.

Currently, there are constraints writing to HDFS. A Flume node can only write to one version of Hadoop. Although Hadoop’s HDFS API has been fairly stable, HDFS clients are only guaranteed to be wire compatible with the same major version of HDFS. Cloudera’s testing used Hadoop HDFS 0.20.x and HDFS 0.18.x. They are API compatible so all that is necessary to switch versions is to swap out the Hadoop jar and restart the node that will write to the other Hadoop version.

You still need to configure this instance of Hadoop so that it talks to the correct HDFS nameNode. You configure the Hadoop client settings (such as pointers to the name node) the same way as with Hadoop dataNodes or worker nodes — modify and use conf/core-site.xml.

By default, Flume checks for a Hadoop jar in /usr/lib/hadoop. If it is not present, it defaults to a jar found in its own lib directory, /usr/lib/flume/ lib.

If a node does not show up on the Master, you should first check to see if the node is running.

# jps | grep FlumeNode

You can also check logs found in /usr/lib/flume/logs/flume-flume-node*.log.

Common errors are error messages or warnings about being unable to contact the masters. This could be due to host misconfiguration, port misconfiguration (35872 is the default heartbeat port), or fire walling problems.

Another possible error is to have a permissions problems with the local machine’s write-ahead log directory. On an out-of-the-box setup, this is in the /tmp/flume/agent directory). If a Flume Node is ever run as a user other than flume, (especially if it was run as root), the directory needs to be either deleted or its contents must have its permissions modified to allow Flume to use it.

Nodes by default start and choose their hostname as their physical node name. Physical nodes names are supposed to be unique. Unexpected results may occur if multiple physical nodes are assigned the same name.

On Ubuntu-based installations, logs are written to /usr/lib/logs/.

There are two limits in Linux — the max number of allowable open files (328838), and max number of allowable open files for a user (default 1024). Sockets are file handles so this limits the number of open TCP connections available.

On Ubuntu, need to add a line to /etc/security/limits.conf

<user> hard nofile 10000

The user should also add the following line to a ~/.bash_profile to raise the limit to the hard limit.

ulimit -n 10000

Flume currently relies on the file system for logging mechanisms. You must make sure that the user running Flume has permissions to write to the specified logging directory.

Currently the default is to write to /tmp/flume. In a production system you should write to a directory that does not automatically get deleted on reboot.

Yes. In the collector sink or dfs sinks you can use the s3n:// or s3:// prefixes to write to S3 buckets.

First you must add some jars to your CLASSPATH in order to support s3 writing. Here is a set that Cloudera has tested (other versions will likely work as well):

s3n uses the native s3 file system and has some limitations on individual file sizes. Files written with this method are compatible with other programs like s3cmd.

s3 writes using an overlay file system must go through Hadoop for file system interaction.