FLUME NG - ADVANCED (must read) - Setting multi-agent flow, Consolidation, Multiplexing the flow

I hope you have read my last blog FLUME NG - BASIC (if not then read it 1st ...)

Here i have described the real time work flow of Flume (Specially when u want flume to do more work for you)

SETTING MULTI-AGENT FLOW

 

In order to flow the data across multiple agents or hops, the sink of the previous agent and source of the current hop need to be avro type with the sink pointing to the hostname (or IP address) and port of the source.

CONFIGURING A MULTI AGENT FLOW

To setup a multi-tier flow, you need to have an avro sink of first hop pointing to avro source of the next hop. This will result in the first Flume agent forwarding events to the next Flume agent. For example, if you are periodically sending files (1 file per event) using avro client to a local Flume agent, then this local agent can forward it to another agent that has the mounted for storage.

## weblog agent config #List sources, sinks and channels in the agent weblog-agent.sources = avro-AppSrv-source weblog-agent.sinks = avro-forward-sink weblog-agent.channels = jdbc-channel #define the flow weblog-agent.sources.avro-AppSrv-source.channels = jdbc-channel weblog-agent.sinks.avro-forward-sink.channel = jdbc-channel #avro sink properties weblog-agent.sources.avro-forward-sink.type = avro weblog-agent.sources.avro-forward-sink.hostname = 10.1.1.100 weblog-agent.sources.avro-forward-sink.port = 10000 #configure other pieces ...

## hdfs-agent config #List sources, sinks and channels in the agent hdfs-agent.sources = avro-collection-source hdfs-agent.sinks = hdfs-sink hdfs-agent.channels = mem-channel #define the flow hdfs-agent.sources.avro-collection-source.channels = mem-channel hdfs-agent.sinks.hdfs-sink.channel = mem-channel #avro sink properties hdfs-agent.sources.avro-collection-source.type = avro hdfs-agent.sources.avro-collection-source.bind = 10.1.1.100 hdfs-agent.sources.avro-collection-source.port = 10000 #configure other pieces ...

 Here we link the avro-forward-sink from weblog-agent to avro-collection-source of  hdfs-agent. This will result in the events coming from the external appserver source eventually getting stored in HDFS.

CONSOLIDATION

A very common scenario in log collection is a large number of log producing clients sending data to a few consumer agents that are attached to the storage subsystem. 

For examples, logs collected from hundreds of web servers sent to a dozen of agents that write to HDFS cluster.

This can be achieved in Flume by configuring a number of first tier agents with an avro sink, all pointing to an avro source of single agent. This source on the second tier agent consolidates the received events into a single channel which is consumed by a sink to its final destination.

MULTIPLEXING THE FLOW

Flume supports multiplexing the event flow to one or more destinations. This is achieved by defining a flow multiplexer that can replicate or selectively route an event to one or more channels.

FAN OUT FLOW

Flume support fanning out the flow from one source to multiple channels. 

There are two modes of fan out, replicating and multiplexing. 

In the replicating flow the event is sent to all the configured channels. 

In case of multiplexing, the event is sent to only a subset of qualifying channels. 

To fan out the flow, one needs to specify a list of channels for a source and the policy for the fanning it out. 

This is done by adding a channel selector that can be replicating or multiplexing. Then further specify the selection rules if its a multiplexer. 

If you dont specify an selector, then by default its replicating.

#List the sources, sinks and channels for the agent <agent>.sources = <Source1> <agent>.sinks = <Sink1> <Sink2> <agent>.channels = <Channel1> <Channel2> #set list of channels for source (separated by space) <agent>.sources.<Source1>.channels = <Channel1> <Channel2> #set channel for sinks <agent>.sinks.<Sink1>.channel = <Channel1> <agent>.sinks.<Sink2>.channel = <Channel2> <agent>.sources.<Source1>.selector.type = replicating

The multiplexing select has a further set of properties to bifurcate the flow. This requires specifying a mapping of an event attribute to a set for channel. The selector checks for each configured attribute in the event header. If it matches the specified value, then that event is sent to all the channels mapped to that value. If theres no match, then the event is sent to set of channels configured as default.

# Mapping for multiplexing selector <agent>.sources.<Source1>.selector.type = multiplexing <agent>.sources.<Source1>.selector.header = <someHeader> <agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1> <agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2> <agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2> ... <agent>.sources.<Source1>.selector.default = <Channel2>

The mapping allows overlapping the channels for each value. The default must be set for a multiplexing select which can also contain any number of channels.

The following example has a single flow that multiplexed to two paths. The agent has a single avro source and two channels linked to two sinks.

#List the sources, sinks and channels in the agent weblog-agent.sources = avro-AppSrv-source1 weblog-agent.sinks = hdfs-Cluster1-sink1 avro-forward-sink2 weblog-agent.channels = mem-channel-1 jdbc-channel-2 # set channels for source weblog-agent.sources.avro-AppSrv-source1.channels = mem-channel-1 jdbc-channel-2 #set channel for sinks weblog-agent.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1 weblog-agent.sinks.avro-forward-sink2.channel = jdbc-channel-2 weblog-agent.sources.avro-AppSrv-source1.selector.type = multiplexing weblog-agent.sources.avro-AppSrv-source1.selector.header = State weblog-agent.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1 weblog-agent.sources.avro-AppSrv-source1.selector.mapping.AZ = jdbc-channel-2 weblog-agent.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 jdbc-channel-2 weblog-agent.sources.avro-AppSrv-source1.selector.default = mem-channel-1

The selector checks for a header called State. If the value is CA then its sent to mem-channel-1, if its AZ then it goes to jdbc-channel-2 or if its NY then both. 

If the State header is not set or doesnt match any of the three, then it goes to mem-channel-1 which is designated as default.

installing configuring and working with FLUME-NG

1st of all i would like to recommend you to use flume NG not flume Og. Below find the difference and architecture of the Flume NG.(Dont mix NG and OG)

What's not compatible between 0.9.x(This is OG - Origional Generation) and 1.x(NG- New Generation) ?

• Events are represented differently and have a different interface
• Source and sink APIs are different
• The RPC mechanisms are different

Architecture

(Here i am trying to get data from web server and collecting it in HDFS)

Flume NG's high level architecture solidifies a few concepts from Flume OG and drastically simplifies others. Flume NG retains Flume OG's general approach to data transfer.
The major components of the system are:

• Event
  An event is a singular unit of data that can be transported by Flume NG.

• Source
  A source of data from which Flume NG receives data.

• Sink
  A sink is the counterpart to the source in that it is a destination for data in Flume NG. Some of the builtin sinks that are included with Flume NG are the Hadoop Distributed File System sink which writes events to HDFS in various ways.

• Channel
  A channel is a conduit for events between a source and a sink. Channels also dictate the durability of event delivery between a source and a sink.

• Source and Sink Runners
  Flume NG uses an internal component called the source or sink runner. The runner is mostly responsible for driving the source or sink and is mostly invisible to the end user.

• Agent
  Flume NG generalizes the notion of an agent. An agent is any physical JVM running Flume NG. Flume OG users should discard previous notions of an agent and mentally connect this term to Flume OG's "physical node." NG no longer uses the physical / logical node terminology from Flume OG. A single NG agent can run any number of sources, sinks, and channels between them.

 Step-by-Step for Flume Ng :

1 . Click on below link to download the Flume NG-1.1.0 or visit download option on https://cwiki.apache.org/FLUME/ to find latest.

http://apache.techartifact.com/mirror/incubator/flume/flume-1.1.0-incubating/apache-flume-1.1.0-incubating-bin.tar.gz

2. Extract it to your workspace (i am doing it in my /home/hadoop/ folder) assume the same in the whole doc.

3. Now got to the Flume folder (in my case it is : /home/hadoop/apache-flume-1.1.0-incubating-bin) and double click on conf folder.

3 files are there.now you run following command in terminal(else you can do it manually)

sudo cp conf/flume-conf.properties.template conf/flume.conf
sudo cp conf/flume-env.sh.template conf/flume-env.sh

now check you have 5 files in the conf folder.

4. open flume.conf in text editor and remove every thing 

add below code (i have added comment below)


# Define a memory channel called ch1 on agent1
agent1.channels.ch1.type = memory

# Here exec1 is source name.
agent1.sources.exec1.channels = ch1
agent1.sources.exec1.type = exec
agent1.sources.exec1.command = tail -F /home/hadoop/as/ash
#in /home/hadoop/as/ash i have kept a text file.

# Define a logger sink that simply logs all events it receives
# and connect it to the other end of the same channel.
# Here HDFS is sink name. 
agent1.sinks.HDFS.channel = ch1
agent1.sinks.HDFS.type = hdfs
agent1.sinks.HDFS.hdfs.path = hdfs://localhost:54310/usr
agent1.sinks.HDFS.hdfs.file.Type = DataStream

# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.
agent1.channels = ch1
#source name can be of anything.(here i have chosen exec1)
agent1.sources = exec1 
#sinkname can be of anything.(here i have chosen HDFS)
agent1.sinks = HDFS

5. Save it and Run athe below line in Terminal and check in UI of Namenode to confirm your file moved to HDFS or not.

bin/flume-ng node --conf ./conf/ -f conf/flume.conf -n agent1

I have written agent1 because in my conf file i have define agent1.If you want to write some other name also it will work.

If you have a different configuration file for this perticular job then please write that in place of  conf/flume.conf. 

Your terminal will look like : 

  
You are done with coping file from local system to HDFS.


5. But remember one thing if you have a different task then you have to make your own configuration file in conf folder.(let say i am creating new file flume.conf1) for a different purpose - i am using here avro source and sink.

Copy this to the new file flume.conf1.

# Define a memory channel called ch1 on agent1
agent1.channels.ch1.type = memory

# Define an Avro source called avro-source1 on agent1 and tell it
# to bind to 0.0.0.0:41414. Connect it to channel ch1.
agent1.sources.avro-source1.channels = ch1
agent1.sources.avro-source1.type = avro
agent1.sources.avro-source1.bind = 0.0.0.0
agent1.sources.avro-source1.port = 41414

# Define a logger sink that simply logs all events it receives
# and connect it to the other end of the same channel.
agent1.sinks.log-sink1.channel = ch1
agent1.sinks.log-sink1.type = logger

# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.

agent1.channels = ch1
agent1.sources = avro-source1
agent1.sinks = log-sink1

7. Now start the avro source

bin/flume-ng node --conf ./conf/ -f conf/flume.conf -n agent1

Here node will start the avro source as agent1 type is avro.

8. Now start another terminal and write below .

bin/flume-ng avro-client --conf conf -H localhost -p 41414 -F /etc/passwd

 You can give your location in the place of /etc/passwd.

9.If your current Terminal showing a message like below....

and old terminal(where u got Avro source started ) showing like below...

 

 Then you have done it correctly.