\section{Introduction}

The deluge of huge data sets such as those provided by sensor
networks, online transactions, and petascale simulation provide 
exciting opportunities for data analytics.  
The scale of the data makes it increasingly difficult to process 
in a reasonable amount of time on isolated machines.
In the near future petascale and exascale simulation will make 
the involvement of secondary storage and disks impractical, driving
the need for infrastructures which provide in-situ analytics and
visualization capabilities~\cite{in-situ}.

% These might be other points to cover:
%%* tight coupling of language, runtime, and workflow methodology
%%* overly complex and difficult to deploy on petascale clusters or 
%%dynamic resources
%%* primarily targeted at post-processing

This has lead to data flow systems emerging
as the standard tool for solving research problems using these vast
datasets.  In typical dataflow systems,
runtimes~\cite{dean2008msd}~\cite{bialecki:hfr}~\cite{isard2007ddd}
define graphs of processes, the edges of the graphs
representing pipes and their vertices representing computation on a
system.  Within these runtimes a new class of
languages~\cite{pike2005idp}~\cite{yu2008dsg}~\cite{olston2008pln} can
be used by researchers to solve "pleasantly parallel" problems
(problems where the individual elements of datasets are considered
to be independent of any other element) more quickly without worrying
about explicit concurrency.

These languages provide automated control flow(typically matched
to the architecture of the underlying runtime) and channels of
communication between systems.  In existing systems, these workflows
and the underlying computation are tightly linked, tying solutions
to a particular runtime, workflow and language.  This creates
difficulties for analytics researchers who wish to draw upon tools
written in many different languages or runtimes which may be available
on several different architectures or operating systems.

Our experiences with existing tools for constructing workflows
for simulation, transformation, analysis and visualization were frustrating.
This was primarily due to the tight coupling of language, runtime, and
workflow tools which proved difficult to integrate with existing
applications.  
We also found many of these systems difficult (if not impossible)
to deploy on petascale clusters, particularly those with dynamic
resources such as clouds where nodes might be added or removed based
on load, failure, or different system priorities.

We observed that UNIX pipes were perfectly designed to allow 
developers to hook together tools written
in different languages and runtimes in ad-hoc fashions.  
This allowed
tool developers to focus on doing one thing well, and enabled code
portability and reuse in ways not originally conceived by the tool
authors.  The UNIX shell incorporated a model for tersely composing
these smaller tools into pipelines (e.g. 'sort $|$ uniq -c'), 
creating a
coherent workflow to solve more complicated problems quickly.  Tools
read from standard input and wrote to standard output, allowing
programs to work together in streams \emph{with no explicit knowledge
of this chaining built into the program itself}.

One to one pipelines such as those used by a typical UNIX shell,
can not be trivially mapped to streaming workflows which incorporate
one-to-many, many-to-many, and many-to-one data flows.  
UNIX shell pipelines are strictly local, so while they can be used in
conjunction with distributed system tools such as ssh, they do not 
themselves naturally facilitiate networked operation.
Additionally,
typical UNIX pipeline tools write data according to buffer boundaries
instead of record boundaries.  
%As \cite{pike2005idp} notes dataflow
%systems need to be able to cleanly separate input streams into
%records and then show that the order of these records is independent.
%By separating input and output into discrete unordered records data
%can be easily distributed and coalesced.

To address these issues we have implemented a prototype shell, which
we call PUSH, using dataflow principles and incorporating extended
pipeline operators to establish distributed workflows ---potentially
running on clusters of machines--- and correlate results. 
In order to scale this approach to large clusters of servers we 
implemented a workload and resource distribution infrastructure which 
incorperated dataflow communication constructs which we call \emph{Brasil}.
%Our
%implementation is based on extending an existing shell, MASH\cite{mashman},
%from which we inherited a rich interpreted scripting language.  It
%treats variables as lists of strings and has no native handling for
%any other data type.  Integer expression handling and other facilities
%are provided by shell commands.  It has native regular expression
%support and it has a novel ability to do declarative shell programming
%through a make like syntax incorporated in the shell itself.
%
%We currently have a working prototype of the PUSH shell, which we
%can use to target local distributed clusters, dynamic clusters built
%using Amazon's EC2 cloud, and large scale clusters such as a Blue
%Gene running the kittyhawk infrastructure.  We are currently in the
%process of evaluating and optimizing performance for a variety of
%application types.
%
%the RC shell\cite{rcpaper} to easier integration into traditional unix
%systems like Linux. This version is simplified and closer to the
%bourne shell. The explicit goal of the new version of Push is
%integration with the Unified Execution Model(UEM)\cite{van-unified} which will
%allow the transparent distribution of processes and the connection
%of their communication channels between machines transparently.
%This work is taking place as part of the HARE project\cite{van2008holistic} 

The rest of this paper is focused on describing the design and implementation 
of Brasil. 
The next section covers some additional details of the PUSH shell and 
describes some of the key design elements of Brasil motivated by it.
Section 3 discusses our prototype implementation in more detail including
lessons learned while attempting to scale the prototype to thousands of cores.
Section 4 contains our evaluation of the overhead of the infrastructure 
when deployed on leadership class high-performance computing environments.
In Section 5 we will conclude by discussing potential improvements to 
address the overheads and explore opportunities for improvement in both 
the design and implementation of our approach.