The streaming topN example

The streaming topN example demonstrates how to use a chain of interceptors to compute a near real-time list of the 10 most popular hashtags from a continuous stream of twitter status updates.

An example agent configuration is located in the resources: /src/main/resources/flume-topn-example.conf.

Cloudera's TwitterSource

A Flume source that pulls status updates from the twitter streaming API and puts events that match the configured search keywords in a Flume channel:

AnalyticsAgent.sources.Twitter.type = com.cloudera.flume.source.TwitterSource
AnalyticsAgent.sources.Twitter.channels = TwitterMem
AnalyticsAgent.sources.Twitter.consumerKey = <required>
AnalyticsAgent.sources.Twitter.consumerSecret = <required>
AnalyticsAgent.sources.Twitter.accessToken = <required>
AnalyticsAgent.sources.Twitter.accessTokenSecret = <required>
AnalyticsAgent.sources.Twitter.keywords = hadoop, big data, analytics, bigdata, cloudera, data science, data scientiest, business intelligence, mapreduce, data warehouse, data warehousing, mahout, hbase, nosql, newsql, businessintelligence, cloudcomputing

You will need to provide your own authentication details for accessing the twitter streaming API. For more information, see twitter's streaming-apis and obtaining access tokens documentation.

For more information on TwitterSource, see Cloudera's GitHub and blog.

HashtagRollingCountInterceptor

A Flume interceptor that counts how many times a hashtag has appeared in twitter status updates in a sliding window style. Hashtags are extracted from a status update, their running counts are incremented, and their new totals are added to the event's header. For example, if the sliding window length is set to 30 minutes and the following status update was seen 50 times in the past 30 minutes, then the emitted event will look like this:

header: {"#Hadoop":"50","#BigData":"50"}
body: {"text":"Follow @ClouderaEng for technical posts, updates, and resources. #Hadoop #BigData"}

HashtagRollingCountInterceptor extends the RollingCountInterceptor class, which provides a generic implementation. Should you want to use a different source, all you have to do is extend RollingCountInterceptor and implement getObjectsToCount().

AnalyticsAgent.sources.Twitter.interceptors.RollingCount.type = org.apache.flume.analytics.twitter.HashtagRollingCountInterceptor$Builder
AnalyticsAgent.sources.Twitter.interceptors.RollingCount.windowLenSec = 600
AnalyticsAgent.sources.Twitter.interceptors.RollingCount.numBuckets = 10

windowLenSec and numBuckets determine the length of the sliding window (how many seconds a hashtag count should be kept) and how granular the counters are. A hashtag's counter is split into buckets, with each bucket representing an equal fraction of time of the overall window length. A hashtag's total count is the sum of its buckets.

Counter increments are added to the head bucket, whilst a separate "reaper" thread advances the window by shifting the head bucket and wiping the tail bucket every (windowLenSec / numBuckets) seconds. Therefore as a hashtag's frequency decreases, its total count also decreases over time. Stale hashtags (those with a total count of zero) are removed from the list to free up memory.

Increasing the number of buckets increases the granularity of the counters (i.e. a higher number of buckets causes the "reaper" thread to run more frequently, which in turn will decrement an objects total count faster as its frequency decreases).

HashtagTopNInterceptor

A Flume interceptor that ranks hashtags by the number of times they have appeared in twitter status updates. HashtagTopNInterceptor consumes events emitted by HashtagRollingCountInterceptor.

HashtagTopNInterceptor extends the TopNInterceptor class, which provides a generic implementation. Should you want to send counters in a different format (e.g. in the body of an event and not in the header), all you have to do is extend TopNInterceptor and implement getCounters().

AnalyticsAgent.sources.Twitter.interceptors.TopN.type = org.apache.flume.analytics.twitter.HashtagTopNInterceptor$Builder
AnalyticsAgent.sources.Twitter.interceptors.TopN.topN = 10

HashtagTopNInterceptor does not emit the topN list itself, it emits the same events as it receives so that the status updates can be stored in HDFS. The topN list is emitted by PeriodicEmissionSource.

PeriodicEmissionSource

A Flume source that connects to HashtagTopNInterceptor and periodically emits the topN list. When HashtagTopNInterceptor initialises it registers itself with InterceptorRegistry:

public void initialize() {
    InterceptorRegistry.register(HashtagTopNInterceptor.class, this);
}

PeriodicEmissionSource then retrieves the interceptor from InterceptorRegistry, connects to it and emits the topN list every 5 seconds.

AnalyticsAgent.sources.TopNSrc.type = org.apache.flume.source.PeriodicEmissionSource
AnalyticsAgent.sources.TopNSrc.channels = TopNMem
AnalyticsAgent.sources.TopNSrc.emitFreqMS = 5000
AnalyticsAgent.sources.TopNSrc.interceptorClass = org.apache.flume.analytics.twitter.HashtagTopNInterceptor

The topN list is emitted in JSON format:

header: {timestamp=1373305825931}
body: {"topN_TS":1373305825931,"topN":[{"#bigdata":28},{"#analytics":9},{"#cloudera":5},{"#freelance":3},{"#cloud":3},{"#wimbledon":3},{"#marketing":2},{"#hadoop":2},{"#stratany2013:":2},{"#oracle":2}]}

Channels and sinks

Both data flows (the status updates coming from TwitterSource and the topN coming from PeriodicEmissionSource) pass through separate memory channels and are written to different locations in HDFS.