\section{Introduction} The deluge of huge data sets such as those provided by sensornetworks, 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, drivingthe need for infrastructures which provide in-situ analytics andvisualization 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 emergingas the standard tool for solving research problems using these vastdatasets.  In typical dataflow systems,runtimes~\cite{dean2008msd}~\cite{bialecki:hfr}~\cite{isard2007ddd}define graphs of processes, the edges of the graphsrepresenting pipes and their vertices representing computation on asystem.  Within these runtimes a new class oflanguages~\cite{pike2005idp}~\cite{yu2008dsg}~\cite{olston2008pln} canbe used by researchers to solve "pleasantly parallel" problems(problems where the individual elements of datasets are consideredto be independent of any other element) more quickly without worryingabout explicit concurrency. These languages provide automated control flow(typically matchedto the architecture of the underlying runtime) and channels ofcommunication between systems.  In existing systems, these workflowsand the underlying computation are tightly linked, tying solutionsto a particular runtime, workflow and language.  This createsdifficulties for analytics researchers who wish to draw upon toolswritten in many different languages or runtimes which may be availableon several different architectures or operating systems. Our experiences with existing tools for constructing workflowsfor simulation, transformation, analysis and visualization were frustrating.This was primarily due to the tight coupling of language, runtime, andworkflow tools which proved difficult to integrate with existingapplications.  We also found many of these systems difficult (if not impossible)to deploy on petascale clusters, particularly those with dynamicresources such as clouds where nodes might be added or removed basedon load, failure, or different system priorities. We observed that UNIX pipes were perfectly designed to allow developers to hook together tools writtenin different languages and runtimes in ad-hoc fashions.  This allowedtool developers to focus on doing one thing well, and enabled codeportability and reuse in ways not originally conceived by the toolauthors.  The UNIX shell incorporated a model for tersely composingthese smaller tools into pipelines (e.g. 'sort $|$ uniq -c'), creating acoherent workflow to solve more complicated problems quickly.  Toolsread from standard input and wrote to standard output, allowingprograms to work together in streams \emph{with no explicit knowledgeof 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 incorporateone-to-many, many-to-many, and many-to-one data flows.  UNIX shell pipelines are strictly local, so while they can be used inconjunction 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 boundariesinstead 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, whichwe call PUSH, using dataflow principles and incorporating extendedpipeline operators to establish distributed workflows ---potentiallyrunning 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 includinglessons 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.