2010 IEEE International Conference on Data Mining Workshops

S4: Distributed Stream Computing Platform

Leonardo Neumeyer Bruce Robbins Anish Nair Anand Kesari

Yahoo! Labs Yahoo! Labs Yahoo! Labs Yahoo! Labs
Santa Clara, CA Santa Clara, CA Santa Clara, CA Santa Clara, CA
neumeyer@[Link] robbins@[Link] anishn@[Link] anands@[Link]

Abstract—S4 is a general-purpose, distributed, scalable, par- minimal overhead and support. The main requirements for a
tially fault-tolerant, pluggable platform that allows program- production environment are scalability (ability to add more
mers to easily develop applications for processing continuous servers to increase throughput with minimal effort) and high
unbounded streams of data. Keyed data events are routed with
affinity to Processing Elements (PEs), which consume the events availability (ability to achieve continuous operation with no
and do one or both of the following: (1) emit one or more events human intervention in the presence of system failures). We
which may be consumed by other PEs, (2) publish results. considered extending the open source Hadoop platform to
The architecture resembles the Actors model [1], providing support computation of unbound streams but we quickly
semantics of encapsulation and location transparency, thus realized that the Hadoop platform was highly optimized for
allowing applications to be massively concurrent while exposing
a simple programming interface to application developers. In batch processing. MapReduce systems typically operate on
this paper, we outline the S4 architecture in detail, describe static data by scheduling batch jobs. In stream computing,
various applications, including real-life deployments. Our de- the paradigm is to have a stream of events that flow into
sign is primarily driven by large scale applications for data the system at a given data rate over which we have no
mining and machine learning in a production environment. control. The processing system must keep up with the
We show that the S4 design is surprisingly flexible and lends
itself to run in large clusters built with commodity hardware. event rate or degrade gracefully by eliminating events, this
is typically called load shedding. The streaming paradigm
Keywords-actors programming model; complex event pro-
dictates a very different architecture than the one used in
cessing; concurrent programming; data processing; distributed
programming; map-reduce; middleware; parallel program- batch processing. Attempting to build a general-purpose
ming; real-time search; software design; stream computing platform for both batch and stream computing would result
in a highly complex system that may end up not being
I. I NTRODUCTION optimal for either task. An example of a MapReduce online
S4 (Simple Scalable Streaming System) is a distributed architecture built as an extension of Hadoop can be found
stream processing engine inspired by the MapReduce model. in [3].
We designed this engine to solve real-world problems in The MapReduce programming model makes it possible to
the context of search applications that use data mining and easily parallelize a number of common batch data processing
machine learning algorithms. Current commercial search tasks and operate in large clusters without worrying about
engines, such as Google, Bing, and Yahoo!, typically provide system issues like failover management [4]. With the surge
organic web results in response to user queries and then of open source projects such as Hadoop [5], adoption of
supplement with textual advertisements that provide revenue the MapReduce programming model has accelerated and is
based on a “cost-per-click” billing model [2]. To render moving from the research labs into real-world applications
the most relevant ads in an optimal position on the page, as diverse as web search, fraud detection, and online dating.
scientists develop algorithms that dynamically estimate the Despite these advances, there is no similar trend for general-
probability of a click on the ad given the context. The context purpose distributed stream computing software. There are
may include user preferences, geographic location, prior various projects and commercial engines ([6], [7], [8], [9],
queries, prior clicks, etc. A major search engine may process [10]), but their use is still restricted to highly specialized
thousands of queries per second, which may include several applications. Amini et. al. [7] provide a review of the various
ads per page. To process user feedback, we developed S4, systems.
a low latency, scalable stream processing engine. The emergence of new applications such as real-time
To facilitate experimentation with online algorithms, we search, high frequency trading, and social networks is push-
envisioned an architecture that could be suitable for both ing the limits of what can be accomplished with traditional
research and production environments. The main require- data processing systems [11]. There is a clear need for
ment for research is to have a high degree of flexibility highly scalable stream computing solutions that can operate
to deploy algorithms to the field very quickly. This makes at high data rates and process massive amounts of data.
it possible to test online algorithms using live traffic with For example, to personalize search advertising, we need to

978-0-7695-4257-7/10 $26.00 © 2010 IEEE 170

DOI 10.1109/ICDMW.2010.172
process thousands of queries per second from millions of modity hardware, and avoiding the use of shared memory
unique users in real-time, which typically involves analyzing across the cluster led us to adopt the Actors model [1]
recent user activity such as queries and clicks. We found that for S4. This model has a set of simple primitives and has
user session features can increase the accuracy of the models been proven to be effective at an industrial scale through its
used to predict ad relevance. This performance improvement use in a variety of frameworks [13]. In S4, computation is
is used to improve the relevance of the ads shown to each performed by Processing Elements (PEs) and messages are
individual user [12]. S4 addresses the need for a general- transmitted between them in the form of data events. The
purpose distributed stream computing platform. state of each PE is inaccessible to other PEs; event emission
It is worth mentioning that many real world systems and consumption is the only mode of interaction between
implement a streaming strategy of partitioning the input data PEs. The framework provides the capability to route events
into fixed-size segments that are processed by a MapReduce to appropriate PEs and to create new instances of PEs. These
platform. The disadvantage of this approach is that the aspects of the design provide the properties of encapsulation
latency is proportional to the length of the segment plus and location transparency.
the overhead required to do the segmentation and initiate The S4 design shares many attributes with IBM’s Stream
the processing jobs. Small segments will reduce latency, Processing Core (SPC) middleware [7]. Both systems are
add overhead, and make it more complex to manage inter- designed for big data and are capable of mining information
segment dependencies (eg. A segment may need information from continuous data streams using user defined operators.
from prior segments). On the other hand, large segments The main differences are in the architectural design. While
would increase latency. The optimal segment size will de- the SPC design is derived from a subscription model, the
pend on the application. Rather than trying to fit a square S4 design is derived from a combination of MapReduce and
peg into a round hole we decided to explore a programming the Actors model. We believe that the S4 design achieves
paradigm that is simple and can operate on data streams in a much greater level of simplicity due to its symmetry; all
real-time. The design goals were as follows: nodes in the cluster are identical and there is no centralized
• Provide a simple Programming Interface for processing control. As we will show, this is accomplished by leveraging
data streams. ZooKeeper [14], a simple and elegant cluster management
• Design a cluster with high availability that can scale service, that can be shared by many systems in the data
using commodity hardware. center.
• Minimize latency by using local memory in each pro-
II. D ESIGN
cessing node and avoiding disk I/O bottlenecks.
• Use a decentralized and symmetric architecture; all We define a stream as a sequence of elements (”events”)
nodes share the same functionality and responsibilities. of the form (K, A) where K, and A are the tuple-valued
There is no central node with specialized responsibil- keys and attributes repectively [9]. Our goal is to design a
ities. This greatly simplifies deployment and mainte- flexible stream computing platform which consumes such
nance. a stream, computes intermediate values, and possibly emits
• Use a pluggable architecture to keep the design as other streams in a distributed computing environment. This
generic and customizable as possible. section contains an example application, followed by de-
• Make the design science friendly, that is, easy to tailed descriptions of the various components of S4.
program and flexible.
A. Example
To simplify the initial S4 design, we made the following
In the example described in Figure 1, input events contain
assumptions:
a document with a quotation in English. The task is to
• Lossy failover is acceptable. Upon a server failure, pro- continuously produce a sorted list of the top K most frequent
cesses are automatically moved to a standby server. The words across all documents with minimal latency. Quote
state of the processes, which is stored in local memory, events are sent to S4 with no key. The QuoteSplitterPE
is lost during the handoff. The state is regenerated using object (PE1) listens for Quote events that have no key.
the input streams. Downstream systems must degrade QuoteSplitterPE is a keyless PE object that processes
gracefully. all Quote events. For each unique word in a document,
• Nodes will not be added to or removed from a running the QuoteSplitterPE object will assign a count and
cluster. emit a new event of type WordEvent, keyed on word.
We found that these requirements are acceptable for WordCountPE objects listen for WordEvent events emit-
most of our applications. In the future, we plan to work ted with key word. For example, the WordCountPE
on solutions for applications where these limitations aren’t object for key word=“said” (PE2) receives all events
acceptable. of type WordEvent keyed on word=“said”. When a
The dual goals of allowing distributed operation on com- WordEvent event for key word=“said” arrives, S4 looks

171
 A keyless event (EV) arrives at PE1 with quote:
up the WordCountPE object using the key word=“said”. If EV Quote “I meant what I said and I said what I meant.”, Dr. Seuss
the WordCountPE object exists, the PE object is called and KEY null
QuoteSplitterPE (PE1) counts unique
VAL quote="I ..." words in Quote and emits events for
the counter is incremented, otherwise a new WordCountPE each word.
object is instantiated. Whenever a WordCountPE object EV WordEvent
increments its counter, it sends the updated count to a EV WordEvent PE1 KEY word="i"
KEY word="said" VAL count=4
SortPE object. The key of the SortPE object is a ran-
VAL count=2 WordCountPE (PE2-4)
dom integer in [1, n], where n is the desired number of keeps total counts for
SortPE objects. Once a WordCountPE object chooses each word across all
EV UpdatedCountEv PE2 PE3 PE4 quotes. Emits an event
a sortID, it uses that sortID for the rest of its existence. KEY sortID=2
any time a count is
updated.
The purpose of using more than one SortPE object is VAL word=said count=9

to better distribute the load across several nodes and/or EV UpdatedCountEv

SortPE (PE5-7) PE5 PE6 PE7
processors. For example, the WordCountPE object for key KEY sortID=9
continuously sorts partial
lists. Emits lists at periodic VAL word="i" count=35
word=“said” sends an UpdatedCountEvent event to a intervals
SortPE object with key sortID=2 (PE5). Each SortPE EV PartialTopKEv MergePE (PE8) combines partial
object updates its top K list as UpdatedCountEvent KEY topk=1234 PE8 TopK lists and outputs final
VAL words={w:cnt} TopK list.
events arrive. Periodically, each SortPE sends its partial
top K lists to a single MergePE object (PE8), using an PE ID PE Name Key Tuple
PE1 QuoteSplitterPE null
arbitrary agreed upon key, in this example topK=1234. The PE2 WordCountPE word="said"
MergePE object merges the partial lists and outputs the PE4 WordCountPE word="i"
latest authoritative top K list. PE5 SortPE sortID=2
PE7 SortPE sortID=9
PE8 MergePE topK=1234
B. Processing Elements
Processing Elements (PEs) are the basic computational Figure 1. Word Count Example
units in S4. Each instance of a PE is uniquely identified
by four components: (1) its functionality as defined by a PE
class and associated configuration, (2) the types of events that lose the count for that word). This memory management
it consumes, (3) the keyed attribute in those events, and (4) strategy is simple but not the most efficient. To maximize
the value of the keyed attribute in events which it consumes. quality of service (QoS), we should ideally remove PE
Every PE consumes exactly those events which correspond objects based on the available system memory and the
to the value on which it is keyed. It may produce output impact the object may have on the overall performance of
events. Note that a PE is instantiated for each value of the the system. We envision a solution where PE objects can
key attribute. This instantiation is performed by the platform. provide the priority or importance of the object. This value is
For example, in the word counting example, WordCountPE application specific, hence the logic should be implemented
is instantiated for each word in the input. When a new word by the application programmer.
is seen in an event, S4 creates a new instance of the PE
corresponding to that word. C. Processing Node
A special class of PEs is the set of keyless PEs, with Processing Nodes (PNs) are the logical hosts to PEs. They
no keyed attribute or value. These PEs consume all events are responsible for listening to events, executing operations
of the type with which they are associated. Keyless PEs on the incoming events, dispatching events with the as-
are typically used at the input layer of an S4 cluster where sistance of the communication layer, and emitting output
events are assigned a key. events (Figure 2). S4 routes each event to PNs based on a
Several PEs are available for standard tasks such as count, hash function of the values of all known keyed attributes in
aggregate, join, and so on. Many tasks can be accomplished that event. A single event may be routed to multiple PNs.
using standard PEs which require no additional coding. The The set of all possible keying attributes is known from the
task is defined using a configuration file. Custom PEs can configuration of the S4 cluster. An event listener in the PN
easily be programmed using the S4 software development passes incoming events to the processing element container
tools. (PEC) which invokes the appropriate PEs in the appropriate
In applications with a large number of unique keys, it order.
may be necessary to remove PE objects over time. Perhaps There is a special type of PE object: the PE prototype. It
the simplest solution is to assign a Time-to-Live (TTL) to has the first three components of its identity (functionality,
each PE object. If no events for that PE object arrive within event type, keyed attribute); the attribute value is unassigned.
a specified period of time, the PE becomes eligible for This object is configured upon initialization and, for any
removal. When system memory is reclaimed, the PE object value V, it is capable of cloning itself to create fully qualified
is removed and prior state is lost (in our example, we would PEs of that class with identical configuration and value V

172
 ZooKeeper [16]. A subset of active nodes are assigned
to particular tasks, while the remaining idle nodes remain
in a pool which can be used as needed (e.g. failover, or
dynamic load balancing). In particular, an idle node may be
registered as a standby for multiple active nodes which may
be assigned to distinct tasks.

III. P ROGRAMMING M ODEL

The high-level programming paradigm is to write generic,

reusable and configurable processing elements that can be
used across various applications. Developers write PEs in
Figure 2. Processing Node the Java programming language. PEs are assembled into
applications using the Spring Framework.
The processing element API is fairly simple and flexible.
for the keyed attribute. This operation is triggered by the Developers essentially implement two primary handlers:
PN once for each unique value of the keyed attribute that it an input event handler processEvent() and an output
encounters. mechanism output(). In addition, developers can define
As a consequence of the above design, all events with some state variables for the PE. processEvent() is
a particular value of a keyed attribute are guaranteed to invoked for each incoming event of the types the PE has
arrive at a particular corresponding PN, and be routed to subscribed to. This method implements the logic for input
the corresponding PE instances withn it. Every keyed PE event handling, typically an update of the internal PE state.
can be mapped to exactly one PN based on the value of the The output() method is an optional method that can be
hash function applied to the value of the keyed attribute of configured to be invoked in a variety of ways. It can be
that PE. Keyless PEs may be instantiated on every PN. invoked either at regular time intervals t, or on receiving
D. Communication Layer n input events. This also means it can be invoked on every
incoming event, in the case where n=1. The output()
The communication layer provides cluster management method implements the output mechanism for the PE, typ-
and automatic failover to standby nodes and maps physical ically to publish internal state of the PE to some external
nodes to logical nodes. It automatically detects hardware system.
failures and accordingly updates the mapping [15].
We demonstrate this with an example. Consider a PE that
Emitters specify only logical nodes when sending mes-
subscribes to an event stream of user search queries, counts
sages. Emitters are unaware of physical nodes or when
instances for each query since the beginning and intermit-
logical nodes are re-mapped due to failures.
tently writes out the counts to an external persister. The
The communication layer API provides bindings in sev-
event stream consists of events of type QueryEvent. The
eral languages (e.g, Java, C++). Legacy systems can use the
class QueryCounterPE implements processEvent()
communication layer API to send input events in a round-
and output() as described in Figure 3. In this case,
robin fashion to nodes in an S4 cluster. These input events
queryCount is the internal PE state variable that holds
are then processed by keyless PEs.
the count for the query corresponding to this PE. Finally,
The communication layer uses a pluggable architecture to
the configuration for the PE is described in Figure 4. In this
select network protocol. Events may be sent with or without
case, the keys property tells us that QueryCounterPE
a guarantee. Control messages may require guaranteed deliv-
subscribes to events of type QueryEvent and is keyed
ery while data may be sent without a guarantee to maximize
on the attribute queryString from the event. The con-
throughput.
figuration ties the PE to a data persistence component
The communication layer uses ZooKeeper [16] to help
externalPersister (this could be an abstraction for
coordinate between nodes in an S4 cluster. ZooKeeper is
a data serving system) and instructs the output() method
an open source subproject of Hadoop maintained. It is a
to be invoked every 10 minutes.
distributed coordination service for distributed applications.
E. Configuration Management System IV. P ERFORMANCE
We envision a management system where human oper-
ators can set up and tear down clusters for S4 tasks, and We introduce a real-world benchmark application, de-
perform other administrative operations. The assignment of scribe how we approached the problem in S4, and present
physical nodes to these S4 task clusters is coordinated using some performance results.

173
 Figure 3. Excerpt from [Link] ID.)
private queryCount = 0; Serve events contain data pertaining to the serve, eg. the
public void processEvent(Event event) serve ID, query, user, ads, etc. The click events, on the
{ other hand, only contain information about the click, and
queryCount ++;
} additionally the serve ID of the serve associated with the
click. To compute CTR at the query-ad level in S4, we need
public void output()
{ to route click and serve events using a key composed of the
String query = (String) [Link]().get(0); query and ad ids. If the click payload doesn’t include query
[Link](query, queryCount);
} and ad information, we need to do a join by serve ID prior to
routing the events with query-ad as the key. Once joined,
the events must pass through a bot filter. Finally, serves and
Figure 4. Excerpt from [Link] clicks are aggregated to compute CTR. A snapshot of the
<bean id="queryCounterPE" event flow is shown in Figure 5.
class="[Link]">
<property name="keys">
<list> EV RawClick
<value>QueryEvent queryString</value> EV RawServe KEY null
</list> KEY null VAL Click Data
</property> VAL Serve Data
<property name="persister" ref="externalPersister">
<property name="outputFrequencyByTimeBoundary" PE1 RouterPE: Routes
value="600"/> keyless input events
</bean> EV Serve EV Click
KEY serve=123 KEY serve=123
VAL Serve Data VAL Click Data
PE2
JoinPE: Joins clicks/
A. Streaming Click-Through Rate Computation serve join using the key
"serve"
User clicks are one of the most valuable user behaviors EV JoinedServe EV JoinedClick
on the web. They provide immediate feedback on the pref- KEY user=Peter KEY user=Peter
erences and engagement of the user which can be used VAL Joined Data VAL Joined Data

to improve user experience by showing the most popular PE3 BotFilterPE: Uses
stateless and stateful
items in more prominent positions. In the case of search rules to filter events
advertising that use a pay-per-click revenue model, publish-
EV FilteredServe EV FilteredClick
ers, agencies, and advertisers determine payments based on KEY q-ad=ipod-78 KEY q-ad=ipod-78
click counts. Click-through rate (CTR) is the ratio of the VAL Joined Data VAL Joined Data
number of clicks divided by the number of ad impressions. PE4 Emit "clean" events using
When sufficient historical data is available, CTR is a good a composite key. (eg.
estimate of the probability that a user will click on an item. query="ipod", adID =
EV Q-Ad-CTR "78")
Precisely because clicks are an invaluable variable for using KEY q-ad=ipod-78
VAL Joined Data CTRPE counts clean serves
in personalization and ranking, it is also subject to click and clicks using a sliding
fraud. Click fraud could be used to manipulate ranking in a window. Computes CTR
Output events are directed and other click metrics
search engine. Click fraud is typically implemented by using to a data server or any
other listener. PE ID PE Name Key Tuple
malicious software running on remote computers (bots) or
PE1 RouterPE null
groups of computers (botnets). Another potential threat is PE2 JoinPE serve=123
impression spam, that is, requests originated by bots. Some PE3 BotFilterPE user="Peter"
of these requests may not be malicious in nature but could PE4 CTRPE q-ad=ipod-78
affect CTR estimations.
In this example (Figure 5), we show how to use S4 to Figure 5. CTR computation
measure CTR in real-time. In the context of search adver-
tising, a user query is processed by the ad engine, returning B. Experimental Setup
a ranked list of ads. In the example, we measure CTR in 1) Online Experiment: We ran the streaming click-
real-time for each query-ad combination. To eliminate click through rate (CTR) on a random sample of live search
and impression noise, we use a set of heuristic rules to traffic. To ensure consistent experience, search engine users
eliminate suspicious serves and clicks. (In this example, a where assigned to the experiment randomly but fixed based
serve corresponds to a user query and is assigned a unique on a hash of their browser cookies. On average, about one
ID. For each serve, a search results page is returned to million searches per day were issued by 250,000 users. The
the user who may or may not click on hyperlinks. Clicks experiment ran for two weeks. The peak observed event rate
associated with that page are tagged with the same unique during this experiment was 1600 events per second. The

174
experimental cluster consisted of 16 servers, each with 4 The source of degradation was due to the fact that the S4
32-bit processors and 2 GB of memory. grid could not process the event stream fast enough at this
The task was to compute click-through rate (CTR) for a rate, hence causing event loss.
query and advertisement combination with very low latency.
The CTR was aggregated over a sliding rectangular window
of 24 hours. This was implemented by dividing the window V. A PPLICATION : O NLINE PARAMETER OPTIMIZATION
into “slots” of 1 hour each and aggregating clicks and In this section, we introduce a real-life practical applica-
impressions for each slot. Subsequently, on the hour, the tion of S4: an online parameter optimization (OPO) system
aggregations from slots within the window were added up [17], for automating tuning of one or more parameters of
and pushed out to a serving system. This method is quite a search advertising system using live traffic. The system
efficient with respect to memory usage, but the trade-off is removes the need for manual tuning and constant human
in update latency. With more memory, we could maintain intervention, while searching a larger parameter space in
finer grained slots, eg., 5 minutes and reduce the update less time than was possible by manual searches. The system
latency. The system, as implemented, provided short term ingests events emitted by the target system (in our case, the
CTR estimates, which were then combined with longer term search advertising system), measures performance, applies
CTR estimates. In the case of a PN failure, we lose the data an adaptation algorithm to determine new parameters, and
in that node, and no short term estimates will be available injects the new parameters back into the target system. This
for the query/advertisement instances that happen to be closed loop approach is similar in principle to traditional
partitioned to that node. In this case, our failover strategy is control systems.
to back off to the long term estimates.
2) Offline Experiment: We also ran an offline stress test, A. Functional design
in which we setup a test cluster of 8 servers, each with 4 We assume the target system (TS) produces output that
64-bit processors and 16 GB of memory. We run 16 PN’s can be represented as a stream and has measurable perfor-
on these machines, with 2 on each. We used real click and mance in the form of a configurable objective function (OF).
serve data from logs of search traffic. We recreated click and We set aside 2 randomly-assigned slices of the output
serve events and computed the real CTR of search queries, stream of the target system as slice1 and slice2.
which we use as gold standard for the accuracy tests. The We require that the target system has the ability to apply
event data consisted of 3 million serves and clicks. different parameter values in each slice and that each slice
can be identified as such in the output stream.
C. Results
The online parameter optimization system (OPO) has 3
The experiment on live traffic showed that we can improve high-level functional components: measurement, comparator
CTR by about 3% with no loss in revenue, primarily through and optimizer.
detecting low quality ads very quickly and filtering them out. 1) Measurement: The measurement component ingests
The offline stress test was aimed at evaluating the per- the slice1 and slice2 streams of the TS and measures
formance of the system under event rates far beyond the the value of OF in each slice. The OF is measured for the
expected operating point. On the test cluster we described duration of a slot. A slot can either be specified in units of
above, we streamed offline generated events through the time of in terms of event counts of the output stream.
S4 grid, in a sequence of runs at progressively increasing 2) Comparator: The comparator component takes as
event rates. At the end of each run, we compare the CTR’s input the measurements produced by the measurement com-
estimated by the system with the true CTR’s computed from ponent and determines if and when there is a statistically
search logs. Figure 6 shows the results from this test. significant difference in performance between the slice1
and slice2 slices. When it determines a significant differ-
Events per second Relative Error in CTR Data Rate ence, it sends a message to the optimizer. If no difference is
2000 0.0% 2.6 Mbps detected after a specified number of measurement slots, the
3644 0.0% 4.9 Mbps slices are declared as equal.
7268 0.2% 9.7 Mbps 3) Optimizer: The optimizer implements the adaptation
10480 0.4% 14.0 Mbps strategy. It takes as input the history of parameter influences,
12432 0.7% 16.6 Mbps including the latest output of the comparator, and outputs
14900 1.5% 19.9 Mbps new parameter values for both the slice1 and slice2
16000 1.7% 21.4 Mbps slices, thus signaling the start of a new “experiment cycle”.
20000 4.2% 26.7 Mbps
Figure 6. Relative Error of CTR Estimate
B. S4 implementation
Figure 7 describes the implementation of the OPO system
The system showed signs of degrading at about 10 Mbps. in S4. The 3 functional components were implemented

175
 EV TSEvent A keyless event (EV) from target system (TS)
KEY null arrives to PE1 with payload containing some state
information about TS, including the sliceID
VAL stateinfo

RouterPE (PE1) re-routes EV to PE2,3 based EV TSEvent

on the sliceID KEY sliceID="slice2"
VAL stateinfo

EV TSEvent PE1
KEY sliceID="slice1" MeasurementPE (PE2,3) computes UF
VAL stateinfo and outputs events keyed on
comparisonID

PE2 PE3
EV MetricsEvent
KEY comparisonID="slice1:slice2"
EV MetricsEvent VAL UF for 'slice2'
KEY comparisonID="slice1:slice2"
VAL UF for 'slice1' ComparatorPE (PE4) aims to determine significant
PE4 difference between 'slice1' and 'slice2', and emits
OptimizeEvent keyed on comparisonID when it does
EV OptimizeEvent
KEY comparisonId="slice1:slice2"
PE ID PE Name Key Tuple
VAL UF history
PE1 RouterPE null
PE2 MeasurementPE sliceID="slice1"
OptimizerPE (PE5) runs adaptation strategy, PE5 PE3 MeasurementPE sliceID="slice2"
maintains history of experiments and
outputs parameters back to TS PE4 ComparatorPE comparisonID="slice1:slice2"
PE5 OptimizerPE comparisonID="slice1:slice2"

Figure 7. OPO implementation in S4

in the form of 3 PEs. The MeasurementPE is keyed about 200,000 users per day. The system ran for two weeks,
on a sliceID, and there is one instance for slice1 trying to optimize a parameter known to have a significant
and slice2 each (for more advanced strategies, we can effect on search engine performance. The optimal parameters
easily extend this to more slices). Measurements of the generated by the OPO system demonstrated reasonable im-
objective function were on slots of fixed time duration. The provements in the primary measures of system performance:
ComparatorPE is keyed on a comparisonID, which revenue by 0.25% and click yield by 1.4%
maps to a pair of slices, in our case, slice1:slice2.
D. Summary
The determination of statistically significant difference be-
tween slices was based on a dependent t-test for paired We demonstrated the design and implementation of an
measurements. It was configured to require a minimum online parameter optimization system using S4. We applied
number of valid measurements. OptimizerPE is keyed on this system to tuning a search advertising system with
slice1:slice2 as well. For the adaptation strategy, we favorable results. This system can be applied to tune any
use a modified version of the Nelder-Mead (aka Amoeba) dynamic system (that satisfies a few requirements described
algorithm [18]: a gradient-free minimization algorithm. The earlier) with tunable parameters.
parameters output by OPO are fed back to the search VI. F UTURE W ORK
advertising serving system. These parameter values control The current system uses static routing, automatic failover
aspects of the serving system, therefore resulting in different via ZooKeeper, but lacks dynamic load balancing and robust
user behavior. live PE migration. We plan to incorporate these features.
C. Results VII. ACKNOWLEDGEMENTS
We ran the OPO system on real traffic slices of a search The authors would like to thank all the colleagues at Ya-
advertising system. The goal was to improve system metrics hoo! who supported and contributed to this project especially
over the current parameter value (tuned using traditional Khaled Elmeleegy, Kishore Gopalakrishna, George Hu, Jon
methods) as much as possible. The objective function was a Malkin, Benjamin Reed, Stefan Schroedl, and Pratyush Seth.
formulaic representation of revenue and user experience on We are also grateful to Yahoo! for making the S4 source
a search engine. The traffic slices were based on partitions code freely available to the community at large under the
of search engine user space: each slice received traffic from Apache License, Version 2.0 [19].

176
 R EFERENCES [14] P. Hunt, M. Konar, F. P. Junqueira, and B. Reed,
“ZooKeeper: wait-free coordination for internet-scale sys-
[1] G. Agha, Actors: A Model of Concurrent Computation in tems,” in USENIXATC’10: Proceedings of the 2010 USENIX
Distributed Systems. Cambridge, MA, USA: MIT Press, conference on USENIX annual technical conference. Berke-
1986. ley, CA, USA: USENIX Association, 2010, pp. 11–11.

[2] B. Edelman, M. Ostrovsky, and M. Schwarz, “Internet ad- [15] K. Gopalakrishna, G. Hu, and P. Seth, “Communication layer
vertising and the generalized second-price auction: Selling using ZooKeeper,” Yahoo! Inc., Tech. Rep., 2009.
billions of dollars worth of keywords,” American Economic
Review, vol. 97, no. 1, pp. 242–259, 2007. [16] Apache ZooKeeper. [Link]

[3] T. Condie, N. Conway, P. Alvaro, J. M. Hellerstein, [17] J. Malkin, S. Schroedl, A. Nair, and L. Neumeyer, “Tuning
K. Elmeleegy, and R. Sears, “MapReduce online,” EECS Hyperparameters on Live Traffic with S4,” in TechPulse 2010:
Department, University of California, Berkeley, Tech. Internal Yahoo! Conference, 2010.
Rep. UCB/EECS-2009-136, Oct 2009. [Online]. Available:
[Link] [18] J. A. Nelder and R. Mead, “A simplex method for function
[Link] minimization,” Computer Journal, vol. 7, pp. 308–313, 1965.

[4] J. Dean and S. Ghemawat, “MapReduce: simplified data [19] The S4 Open Source Project. [Link]
processing on large clusters,” Commun. ACM, vol. 51, no. 1,
pp. 107–113, 2008.

[5] Apache Hadoop. [Link]

[6] D. C. Luckham, The Power of Events: An Introduction to

Complex Event Processing in Distributed Enterprise Systems.
Boston, MA, USA: Addison-Wesley Longman Publishing
Co., Inc., 2001.

[7] L. Amini, H. Andrade, R. Bhagwan, F. Eskesen, R. King,

P. Selo, Y. Park, and C. Venkatramani, “SPC: a distributed,
scalable platform for data mining,” in DMSSP ’06: Pro-
ceedings of the 4th international workshop on Data mining
standards, services and platforms. New York, NY, USA:
ACM, 2006, pp. 27–37.

[8] D. J. Abadi, Y. Ahmad, M. Balazinska, U. Çetintemel,

M. Cherniack, J.-H. Hwang, W. Lindner, A. Maskey,
A. Rasin, E. Ryvkina, N. Tatbul, Y. Xing, and S. B. Zdonik,
“The Design of the Borealis Stream Processing Engine,” in
CIDR, 2005, pp. 277–289.

[9] D. J. Abadi, D. Carney, U. Çetintemel, M. Cherniack, C. Con-

vey, S. Lee, M. Stonebraker, N. Tatbul, and S. Zdonik,
“Aurora: A new model and architecture for data stream
management,” The VLDB Journal, vol. 12, no. 2, pp. 120–
139, 2003.

[10] Streambase. [Link]

[11] M. Stonebraker, U. Çetintemel, and S. Zdonik, “The 8 re-

quirements of real-time stream processing,” SIGMOD Rec.,
vol. 34, no. 4, pp. 42–47, 2005.

[12] S. Schroedl, A. Kesari, and L. Neumeyer, “Personalized ad

placement in web search,” in ADKDD ’10: Proceedings of
the 4th Annual International Workshop on Data Mining and
Audience Intelligence for Online Advertising, 2010.

[13] R. K. Karmani, A. Shali, and G. Agha, “Actor frameworks

for the JVM platform: a comparative analysis,” in PPPJ ’09:
Proceedings of the 7th International Conference on Principles
and Practice of Programming in Java. New York, NY, USA:
ACM, 2009, pp. 11–20.

177