DEIM Forum 2013 A5-5

QueueLinker: A Framework for Parallel Distributed Processing of Data Streams
Takanori UEDA†,‡ Koh SATOH‡

Daichi SUZUKI‡ Sayaka AKIOKA† and Hayato YAMANA††,‡‡†Information Technology Research Organization, Waseda University, 3-4-1 Okubo, Shinjuku-ku, Tokyo, 169-8555 Japan‡Graduate School of Science and Engineering, Waseda University, 3-4-1 Okubo, Shinjuku-ku, Tokyo, 169-8555 Japan††Faculty of Science and Engineering, Waseda University, 3-4-1 Okubo, Shinjuku-ku, Tokyo, 169-8555 Japan‡‡National Institute of Informatics, 2-1-2 Hitotsubashi, Chiyoda-ku, Tokyo, 101-8430 Japan
E-mail:{t-ueda, kohsatoh, dsuzuki, akioka, yamana}@yama.info.waseda.ac.jp

Abstract With the development of computer systems, many more devices are being connected to the network andgenerating ‘data stream.’ Analyzing data streams in real-time offers valuable information about human activities andcontributes to many information services. QueueLinker enables programmers to build data stream processing applications byimplementing application modules that use a producer–consumer model, and specifying a logical directed graph representing
the data-flow connections between these modules. Each module is automatically instantiated and executed in parallelaccording to the logical directed graph. The data generated by a module is automatically serialized and transferred to othermodules across the network, relieving the programmer of complex multi-threading and communication implementations. In
addition, data parallel model of QueueLinker helps the developers to realize parallel processing without concurrency control.
Keyword Parallel Distributed Processing, Data Stream, Data Parallel, Producer-Consumer
success in processing big data. Now, it is used to process

1. Introduction
Recentcomputerparallel-distributed

computingWith the internet now a common component of ouravailable for everyone. As CPU manufacturers haveinfrastructure and the connection of many mobile devices

decided to increase the number of cores on a CPU insteadto the network, our lifestyles have changed dramaticallyof increasing individual core frequency, the resulting

since MapReduce was introduced in 2004. A constant

plummet in personal computer prices has allowed for the

stream of information is now posted to social network

ownership of shared-nothing clusters at a low cost. In

services with 400 million Twitter tweets generatedorder to take full advantage of recent computer hardware

worldwide per day 2 and the number of digital sensors

performance, developers must now become familiar withgenerating

parallel-distributed computing; however, most developers

information has increased considerably. The analysis ofwant to avoid implementing concurrency control anddata streams in real-time is important in many applications

network communication procedures, as these have provenincluding—among others—social analysis, stock market

difficult to program.

predicting, and healthcare monitoring.

have

made

in

variety of big data.

commodity

hardwareadvancements‘data

Advances

These factors have led to the development ofinstreams’

thatinformation

containvaluabletechnology

have

as

corresponded to an increase in the number of mobile

Google MapReduce [1] and Dryad [2]. Apache Hadoop 1 ,devices and sensors, which in turn has resulted in thean open source implementation of MapReduce, is now

generation

widely used for processing big data. It realizes distributed

processing

computinguser

parallel-distributed computing. Processing data streams inthereal-time presents programming difficulties owing to the

MapReduce programming context, developers are tasked

fact that applications must handle multiple data sources,

with identifying and defining the data parallelism of their

wait for data arrival, and receive data via the network.

applications, after which the framework can scale out data

Because it is difficult for many programmers to consider

processing

available

these factors, it is vital that a framework for real-time,computation nodes. The effectiveness of this kind ofparallel-distributed data-stream processing be developed.

parallel-distributedfor

implementation

bycomputation

large
of

data

network

simplyframeworks

without

requiring

communications.

distributing

such

job

to

In

parallelism is a major reason why Hadoop enjoys broad
1

Welcome to Apache Hadoop!, http://hadoop.apache.org/

2

of

large

numbers

numerous

large

of

data

streams,streamsandrequiresTwitter hits 400 million tweets per day, mostly mobile, http://news.cnet.com/8301-1023_3-57448388-93/twitt
er-hits-400-million-tweets-per-day-mostly-mobile/

With

this

in

mind,

we

QueueLinker,aframework

data-stream

processing.

have

been

developing

interface is designed for storing data in secondary storage

paralleldistributedor visualizing application window to a user.

QueueLinker

enablesfor2.1. Push Module Interface

programmers to build data stream processing applications

Figure 1 shows pseudo code for a push module. In

by implementing application modules that use a producer–

this example, the module has two input queues and an

consumer model, and specifying a logical directed graphoutput queue. An item transferred from another module isrepresenting the data-flow connections between thesepassed through the variable ‘item’. The queue that themodules. Each module is automatically instantiated anditem was put into is identified by the variable ‘queueId’.

executed in parallel according to the logical directed

The module processes an input item and then returns agraph. The data generated by a module is automaticallystring as a result. Processing can differ depending on theserialized and transferred to other modules across the

input queue the item was put into. If the module returns

network,complexnull, QueueLinker sends no data to the next module. This

multi-threading and communication implementations. In

mechanism helps us implement modules for filtering ofaddition, data parallel model of QueueLinker helps thedata.

relievingdevelopers

to

the

realize

programmer

parallelofprocessing

without

concurrency control.

BecauseQueueLinker

uses

a

push

threadunit(described in 5.1) to execute multiple push modules, aThis paper is organized as follows. Section 2

push module cannot use an infinite loop or perform

describes the programming interface of QueueLinker.

blocking operations like file I/O. If a push module does

Sectionnot return, other push modules will not be executed.

3

describes

the

data

parallel

model

of

QueueLinker. Section 4 describes how to define a logicaldirected graph representing data-flow between modules.
Section5

describes

thesoftwarearchitecture

2.2. Pull Module Interface
Figure 2 shows pseudo code for a pull module. In thisofexample, the module pulls an item from an input queue

QueueLinker. Section 6 explains several applications

and then outputs a string toward an output queue.

implemented by using QueueLinker. Section 7 concludes

QueueLinker assigns a dedicated thread to each instancethis paper.

of the pull module. Thus, a pull module implementation

2. Programming Interface of QueueLinker
In QueueLinker, a module is a software processing
unit implemented according to the producer–consumer
model

commonlymulti-threaded

described

programming.

indesignUnderpatterns

theforproducer–consumer model, a module processes input item(s) andgenerates a result. Modules communicate with each otherusing queues. A module sending an item to another module

public class ExamplePushModule extends PushModule<String, String>{@Override
public String execute(String item, int queueId){
if (queueId == 0)
return "Waseda";
else if (queueId == 1)
return "University";
return null;}}

Figure 1

puts the item into the input queue of the destination

Pseudo Code of a Push Module

module. Because modules are not meant to share theirinternal states, an arbitrary number of threads can be

public class ExamplePullModule extends PullModule<String, String> {assigned to each, and they can be executed in parallel on@Override
public void execute(InputStaff<String> inputStaff,
OutputStaff<String> outputStaff,
QueueLinkerService service){
InputQueue<String> input = inputStaff.getDefaultInputQueue();
OutputQueue<String> output = outputStaff.getDefaultOutputQueue();multiple computers to achieve greater speed.
QueueLinker presents a Java API for constructing
producer–consumer modules of four major types: pushwhile (!service.stopRequested()) {try {String item = input.take();modules, pull modules, source modules and sink modules.
Push module interface is designed for non-blocking/* Something to do and generate newStr */operations such as filtering and arithmetic operations. Pull

output.put(newStr);} catch (InterruptedException e) {}module interface is designed for blocking operations suchas file I/O and receiving data from external data sources.
Source module interface is designed for providing data toother modules from external data sources. Sink module}}

Figure 2

Pseudo Code of a Pull Modulecan make use of an infinite loop, which is useful for

created for each module and a thread is assigned to theimplementing

module. Thus, “Tweets Parse” and “Word Count” run on

blocking

operations.

To

do

this,

the

blocking operation is simply written inside an infinite

different threads.

loop in a pull module.

2.3. Source Module Interface

In the general producer–consumer model, an instanceis executed by multiple threads. Thus, modules must be

Figure 3 shows pseudo code for a source module. A

implemented for thread-safety by using concurrency

source module has no input queue and has only one output

control to avoid inconsistency. Concurrency control can bequeue. A data source is typically used to provide data toan especially difficult task, and even when it performed

other modules from external data sources. For example, aproperly, the possibility of lock contention will increase

data source may leverage the Twitter API to feed tweets towith the number of threads executing the instance. To

the system. QueueLinker can manage multiple data

solve this problem, QueueLinker uses data parallel

sources and automatically duplicate the data if multiple

execution with a hash partitioning technique to ensure thatmodules need the data from a single data source.

each instance of a module is executed by only one

2.4. Sink Module Interface

dedicated thread, allowing the programmer to implement

QueueLinker also provides an interface for data sink.

modules without concurrency control.

A sink module has one input queue and no output queue,

For example, in pattern (2) of the figure, a word is

and is typically used to store data in secondary storage ortransferred to one of the two “Word Count” instancespresent that data in visualize window to a user. Figure 4

depending on the hash value. QueueLinker automaticallyshows pseudo code for a sink module.

transfers words that have the same hash value to the same
instance.

3. Data Parallel Execution
This section describes how QueueLinker executes
modules in a parallel-distributed way. Figure 5 shows anexample consisting of three execution patterns for two
modules, “Tweets Parse” and “Word Count”. A rectangle
with a dashed line represents a computer and eachrectangle represents a thread executing a module instance.
The “Tweets Parse” module parses a tweet and outputs theextracted words from the tweet. The “Word Count” module
counts the number of appearance of each word.
In pattern (1) of the figure, only one instance ispublic class ExampleSourceModule extends SourceModule<String>{@Override
public void execute(OutputStaff<String> staff, QueueLinkerService service){
OutputQueue<String> output = staff.getDefaultOutputQueue();
while (!service.stopRequested()) {try {output.put("Output Something");} catch (InterruptedException e) {}}Thismechanism

eliminates

the

need

is executed by only one thread. Note that developers mustspecify modules to be executed by this mechanism when
they define an application; QueueLinker cannot inferautomatically which modules can be parallelized in thisway.
Modules executed in this way do not share theirinternal states with other modules and only communicate
with other modules via queues. Thus, they can be run on
any computer. Pattern (3) in the figure shows an exampleof parallel-distributed execution on three computers.
QueueLinker

automaticallytransfers

items

(1)
Pipeline
Execution

Morph
Analyzer

Parallel
Execution

Morph
Analyzer

(2)

}

Word
Count

Word
Count
Word
Count

}

Figure 3

Pseudo Code of a Source Module

public class ExampleSinkModule extends SinkModule<String> {@Override
public void execute(String input, int queueId) {/* Something to do */}}Figure 4

Pseudo Code of a Sink Modulefor

concurrency control of the module because each instance(3)
Parallel
Distributed
Execution

Morph
Analyzer

A Computer

Word
Count
Word
Count
Word
Count
Word
Count

A Thread executing a module instance

Figure 5

Data Parallel Execution Model ofQueueLinker

betweenmodules, developers do not need to implement network

realizes parallel distributed execution by automaticallycommunication procedures.

instantiating the modules on available computers andtransferring

4. Application

Definition

Using

a

Logical

data

items

betweenthemodules.

The

programmer does not need to know whether transfers
between modules require network communication.

Directed Graph
A QueueLinker user can build an application by
specifying connections between modules. The directed

4.1. Switcher and Virtual Module
QueueLinker

providesamechanism

calledagraph representing these connections is called a ‘logical‘switcher ’ for choosing a destination module based on the

directed graph’. Figure 6 shows a logical directed graphresult data a module produces. It also offers a mechanism,

for the proposed Web crawler described in [5]. Each∈node

is indicated by a rectangle and represents acalled a ‘virtual module’ that allows modules to be reusedin different data flows.

is indicated by a line and

The logical directed graph in Figure 9 includes arepresents a connection between two modules. A node inswitcher, indicated by a circle containing the number ofthe graph is called a ‘logical vertex’ and an edge is called

destination modules. Figure 10 provides pseudo code for aa ‘logical edge’.

switcher. The switcher returns an integer specifying the∈module; each edge

Users can specify the parallel execution mode ofdestination module. QueueLinker will send an item to amodules as well as connection settings. Figure 7 provides

module based on this number. For example, in Figure 9, an

pseudo code describing a logical directed graph for the

output of module A is sent to B if the switcher returns 0,

application shown in Figure 8. In the code, the execution

or to C if the switcher returns 1. Thus, the switcher

mode of the ‘Word Count’ module is set to data parallel

provides control over data routing independent of module

mode by hash partitioning the three instances. The

implementation.

function of the module is to count the number of

The logical directed graph also includes a virtualappearances of each word in tweets. In this case,module, indicated by a rectangle with a dashed line. In theQueueLinker instantiates three instances on different

logical directed graph, outputs of module B are sent tothreads and transfers each string output from the ‘Morphvirtual module A. The virtual module is executed using theAnalyzer ’ module to the correct instance based on the hashsame instance and the thread of module A shown at the far

value of the string. Note that the code does not specify

left of the figure, but outputs of the virtual module aredata parallel mode for the ‘Morph Analyzer ’ module. In

sent to module D. Thus, the virtual module makes it

this case, QueueLinker transfers each tweet to one of thepossible to reuse a module in a different data flow. For

two instances in round-robin fashion.

example, the Web crawler described in [5] uses multiple

As described above, when QueueLinker accepts

switchers and virtual modules. The logical directed graphmodule implementations and a logical directed graph, it(0) robotsFlag = 1

(k)

(m)

robots.txt
Processor

Notification of
Downloading Completion

Seeder(1) The IP is already resolved
robotsFlag = 0, 2(j)(i) Host
Data
Cache

(i) Host
Data
Cache

DomainName(2) IP Resolver

3

Check
the Cache

Notification of
Downloading Completion(i) Host
Data
Cache

Record whether
robots.txt
exists or not(0)
robotsFlag = 0(1)
robotsFlag
=1

2

(k)

robots.txt
Processor

(b)

(a)

Scheduler

Cache
the IP

Unknown(c)robots.txt
Downloader(h)Duplicated
URL Checker

Scheduler
Timer

(0)
robotsFlag
=0

(1)
2
robotsFlag
= 1,2

(2)
robotsFlag=2(g)

Explicit
URL Filter

graph.addLogicalEdge(twitter,morphAnalyzer);graph.addLogicalEdge(morphAnalyzer, wordCount);
graph.addLogicalEdge(wordCount,ui);QueueLinkerClient client = QueueLinkerClientFactory.getClient();
QueueLinkerJob job = new QueueLinkerJob(graph);
JobHandle handle = client.startJob(job);

Figure 7

3

Pseudo Code Describing an Applicationand Submitting the Job(i) Host
Data
Cache

(0)
robotsFlag=0(f)

URL Format
Filter(1)
robotsFlag = 1

Compilerobots.txt

Not Downloadable

Figure 6(d)
Downloader

LogicalGraph graph= new LogicalGraph();LogicalVertex twitter= graph.addLogicalVertex(TwitterDataSource.class);
LogicalVertex morphAnalyzer = graph.addLogicalVertex(MorphAnalyzer.class, 2);
LogicalVertex wordCount= graph.addLogicalVertex(WordCount.class, 3, PMode.Hash);
LogicalVertex ui= graph.addLogicalVertex(UI.class);(k)robots.txt
Processor

2(l)Data Store(e)HTML Parser

Downloadable

An Example of a Logical Directed Graph(Our Proposed Web Crawler Described in [5])Figure 8

A Logical Directed Graph Described by
the Code in Figure 7

of the crawler is shown in Figure 6. Despite being simple

thread input queue, and the CPU core that a thread unit

mechanisms,

areruns on can be controlled using system calls like

indispensable for describing a complex logical directedsched_setaffinity on Linux. ‘Busy wait’ is important forgraph efficiently.

achieving low latency execution of continuous queries

theswitcherandvirtualmodule(described in [4]). In addition, a push thread unit has amechanism for collecting statistics on operator execution,

5. Software ArchitectureThis section provides an overview of the software

such as the number of input/output items to/from, and thearchitecture of QueueLinker. QueueLinker uses several

total CPU time consumed by, each operator. Note that thesoftware mechanisms to execute modules and control

scheduler and the router in a push thread unit are only

execution.

used by that thread unit, and thus do not require any

5.1. Push Thread Unit

concurrency control.

A push thread unit is designed to execute multiple

A number of optimizations should be considered forpush modules and sink modules as illustrated in Figure 11.

the thread local scheduler, since the strategy of the

A push thread unit has a “thread local scheduler” and ascheduler will affect the processing latency, throughput“thread local router” for handling multiple modules. An

and memory consumption of applications. QueueLinker

item to be processed by a module in a thread unit is put

normally uses a FIFO scheduler, but other algorithms,

into the “thread input queue”. The thread unit fetches the

such as Chain [3], can be substituted.

item from the queue, and the thread local router, according

5.2. Pull Thread Unit

to the logical directed graph, determines which module

A pull thread unit executes only one pull or source

will process the item. It then sends the item to the input

module. It must execute that pull module on a single

queue of that module. The thread local scheduler thenthread, since a pull thread unit may contain an infinite

chooses an executable module and executes it. When an

loop (as described in 2.2) and may therefore refuse toitem is produced from the executed module, the thread

yield to other modules. Like the push thread unit, a pull

local router determines the destination of that item. If thethread unit has a thread local router to determine thedestination module runs in data parallel mode, the local

transfer route of each result, but unlike the push threadrouter calculates the hash value of the output item andunit, it does not have a local scheduler, since it does nottransfers

theexecute multiple modules. Other mechanisms of the pull

destination module is running in a thread unit on a remotethread unit are nearly identical to those of the push

computer, QueueLinker transfers the item to that computer,module unit, and are therefore omitted.

using a thread unit dedicated for network communication.

5.3. Master and Worker Server

it

to

the

appropriate

threadunit.

If

A thread unit can ‘busy wait’ for items to arrive in theQueueLinker uses a master server to manage allcomputation nodes, or ‘worker severs’. It accepts job

Virtual Module

servers, which in turn manage thread units. QueueLinker

B

A

E

Switcher

C

D

F

(1)

A Virtual Module and a Switcher

Figure 10

Pseudo Code of a Switcher

Thread
Input
Queue

Mod 1

Mod 2

Thread
Input
Queue

(4)

Mod 3

Mod n

Queue

public class SwitcherExample extends FlowSwitcherModule<String>{@Override
public int execute(String input) {if (input.length() % 2 == 0)
return 0;
else
return 1;}}(3)

RemoteComputer(2)

Thread Local Router

Queue

Figure 9

2

Queue

A

1

Network(4)

Push Thread Unit

Queue

0

requests from clients and sends commands to the worker

Thread Local Scheduler
CommandQueue

RunnableMod
1

WaitingMod
1

Figure 11

Mod
2

Mod
3

Mod
n

A Push Thread Unit

ThreadUnituses ZooKeeper 3 to communicate among master server,

computational resources. Another positive effect of theworker servers, and clients.

QueueLinker model is that it enables us to analyze Web

Figure 12 shows a worker server and its constituent

data in real-time using the flow of data between modules.

thread units. A worker server has a worker local scheduler

We can also easily customize the crawler by changing thethat collects operator statistics from thread units, such as

module implementation.

the number of input/output items to/from, and the total

QueueLinker enables monitoring of the crawling

CPU time consumed by, each operator. The proposed

progress. Figure 13 shows a visualization of a crawling formethodthe Internet with 4 computers. Statistics on the number of

in

[4]

can

be

implemented

by

using

this

items processed by each module and the amount of

mechanism.

resources consumed by each module can be obtained withthe help of the QueueLinker statistics mechanism.

6. QueueLinker Applications
QueueLinker

to6.2. Web Analytics Application

can

be

used

executewidelyapplications

such

as

Web

crawlers,

Web

analytics

As other applications, we have been developing Web

applications

andcontinuous

queries.

This

sectiondata analytics applications. One of those applications is

describes the overview of these applications.

TV chatter extraction from Twitter. People now post their

6.1. A Parallel Distributed Web Crawler

opinions about TV programs for Twitter when they are

QueueLinker can be used to implement a high-speed,

watching TV. Such tweets can be used for audience

parallel-distributed Web crawler [5]. As Web crawlers

analysis. We have been implementing such application bymust collect Web data while performing tasks such as theusing QueueLinker and Hadoop to extract them in

detection

real-time as shown in Figure 14.

ofcrawled

URLs

and

the

preventionofconsecutive access to a certain Web server, parallel and6.3. Continuous Queries

distributed crawling is necessary in achieving high-speedThe

crawling of the extremely high number of URLs that exist

implementation

mechanismon the Web.

“continuous query [6],” which has been studied in theofofQueueLinker

usefulsupports

applications,theincluding

The proposed Web crawler consists of QueueLinker

field of database science since the early 2000s. A

modules. The logical directed graph of the crawler isrelational continuous query can usually be described using

shown in Figure 6. It realizes polite crawling by ensuring

an SQL-like language [7] and compiled to a plan treethat access to a certain Web server does not occur moreconsisting of relational algebra operators. When a tuple

than once in a given interval. Every module is designed

arrives to the system, the tuple is pushed into a leaf of thealong the data parallel model of QueueLinker, and thus

plan tree and the plan tree generates the result of theevery module can run on any number of computers and any

query.

number of threads. In other words, the crawler can assign

QueueLinker can also be used to execute continuous

any computational resources to each module independently.
In addition, the crawler uses data structures that are
temporally and spatially efficient, which allows us to
crawl a large number of URLs with a small amount ofThreadInput
Queue

ThreadUnit

ThreadInput
Queue

ThreadUnit

ThreadInput
Queue

ThreadUnit

ThreadInput
Queue

ThreadUnit

ThreadInput
Queue

ThreadUnit

ThreadInput
Queue

ThreadUnit

Worker Local Scheduler

Figure 12

Thread Units on a Worker Server

Figure 13

Crawling for the Internet with 4
Computers (Each Computer is3

Apache ZooKeeper – Home,
http://zookeeper.apache.org/

Represented as a Yellow Rectangle)

query by implementing a relational algebra operator as amodule, and the corresponding plan tree can be described[4]

using a logical directed graph. Thus, QueueLinker isusefulin

executingcontinuous

query

in

a

parallel-distributed environment.

[5]

7. Conclusion
This paper described the proposed QueueLinker
framework. QueueLinker adopts a producer–consumer
programming

model,

andacceptsaJava

moduleimplementation along with a logical directed graph. Basedon these, it automatically executes each module in thegraph in parallel-distributed manner. Data generated by amodule is automatically serialized and transferred to othermodules across the computational network, even if they
are running on other computers. Programmers do not need
to

write

multi-threaded

programs

or

network

communication procedures.

Acknowledgement
This research was supported by grant from JST“Multimedia

Web

AnalysisFramework

towards

Development of Social Analysis Software”

References[1] J. Dean and S. Ghemawat, “MapReduce: Simplified
Data Processing on Large Clusters,” In Proceedingsof the 6th Symposium on Operating Systems Design
and Implementation (OSDI), pp.137-250, San
Francisco, US-CA, Dec. 2004.
[2] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly,“Dryad: Distributed Data-Parallel Programs from
Sequential Building Blocks,” In Proceedings of the
2nd ACM SIGOPS/EuroSys European Conference on
Computer Systems (EuroSys), pp.59-72, Lisbon,
Portugal, Mar. 2007.
[3] B. Babcock, S. Babu, M. Datar, R. Motwani, and D.
Thomas, “Operator scheduling in data stream
systems,” The VLDB Journal, Vol.13, pp.333-353,

Figure 14

TV Chatter Extraction

[6]

[7]

2004.
T. Ueda, S. Akioka and H. Yamana, “Low Latency
Data Stream Processing on Multi-core CPU
Environments,” IEICE Transactions on Informationand Systems, Vol. 96, no. 5, May 2013 (Japanese, To
Appear).
T. Ueda, K. Satoh, D. Suzuki, K. Uchida, K.
Morimoto, S. Akioka, H. Yamana, “A Parallel
Distributed
Web
Crawler
Consistingof
Producer-Consumer Modules,” IPSJ Transaction
Database, vol. 57, Mar. 2013 (Japanese, To Appear).
S. Chakravarthy and Q. Jiang. Stream Data
Processing: A Quality of Service Perspective.
Springer, 2009.
A. Arasu, S. Babu, and J. Widom, “The CQL
continuous query language: semantic foundations andquery execution,” The VLDB Journal, vol.15,
pp.121-142, Jun. 2006.