WIXLIB

job graphation of code and static data for the remote Dryad vertices; and (c) the generation of serialization code for therequired data types. Section 4 describes these steps indetail.data planeFiles, TCP, FIFONSVVVPDPDPDcontrol planeJob managerStep 4. DryadLINQ invokes a custom, DryadLINQspecific, Dryad job manager. The job manager may beexecuted behind a cluster firewall.Step 5. The job manager creates the job graph using theplan created in Step 3. It schedules and spawns the vertices as resources become available. See Figure 1.clusterFigure 1: Dryad system architecture. NS is the name server whichmaintains the cluster membership. The job manager is responsiblefor spawning vertices (V) on available computers with the help of aremote-execution and monitoring daemon (PD). Vertices exchange datathrough files, TCP pipes, or shared-memory channels. The grey shapeindicates the vertices in the job that are currently running and the correspondence with the job execution graph.2.1DryadLINQ Execution OverviewFigure 2 shows the flow of execution when a program isexecuted by DryadLINQ.Client oreach.NET 4)ExecplanInputtablesData center(5)DryadExecution(8)OutputTables(7)(6)Figure 2: LINQ-expression execution in DryadLINQ.Step 1. A .NET user application runs. It creates aDryadLINQ expression object. Because of LINQ’s deferred evaluation (described in Section 3), the actual execution of the expression has not occurred.Step 2. The application calls ToDryadTable triggering a data-parallel execution. The expression object ishanded to DryadLINQ.Step 3. DryadLINQ compiles the LINQ expression intoa distributed Dryad execution plan. It performs: (a) thedecomposition of the expression into subexpressions,each to be run in a separate Dryad vertex; (b) the gener-USENIX AssociationStep 7. When the Dryad job completes successfully itwrites the data to the output table(s).Step 8. The job manager process terminates, and it returns control back to DryadLINQ. DryadLINQ createsthe local DryadTable objects encapsulating the outputs of the execution. These objects may be used asinputs to subsequent expressions in the user program.Data objects within a DryadTable output are fetchedto the local context only if explicitly dereferenced.Step 9. Control returns to the user application. The iterator interface over a DryadTable allows the user toread its contents as .NET objects.Step 10. The application may generate subsequentDryadLINQ expressions, to be executed by a repetitionof Steps 2–9.3ResultsJMStep 6. Each Dryad vertex executes a vertex-specificprogram (created in Step 3b).Programming with DryadLINQIn this section we highlight some particularly useful anddistinctive aspects of DryadLINQ. More details on theprogramming model may be found in LINQ languagereference [2] and materials on the DryadLINQ projectwebsite [1] including a language tutorial. A companiontechnical report [38] contains listings of some of the sample programs described below.3.1LINQThe term LINQ [2] refers to a set of .NET constructsfor manipulating sets and sequences of data items. Wedescribe it here as it applies to C# but DryadLINQ programs have been written in other .NET languages including F#. The power and extensibility of LINQ derive froma set of design choices that allow the programmer to express complex computations over datasets while givingthe runtime great leeway to decide how these computations should be implemented.The base type for a LINQ collection is IEnumerable T . From a programmer’s perspective, this is8th USENIX Symposium on Operating Systems Design and Implementation3

an abstract dataset of objects of type T that is accessed using an iterator interface. LINQ also definesthe IQueryable T interface which is a subtype ofIEnumerable T and represents an (unevaluated) expression constructed by combining LINQ datasets using LINQ operators. We need make only two observations about these types: (a) in general the programmer neither knows nor cares what concrete type implements any given dataset’s IEnumerable interface; and(b) DryadLINQ composes all LINQ expressions intoIQueryable objects and defers evaluation until the resultis needed, at which point the expression graph within theIQueryable is optimized and executed in its entirety onthe cluster. Any IQueryable object can be used as anargument to multiple operators, allowing efficient re-useof common subexpressions.LINQ expressions are statically strongly typedthrough use of nested generics, although the compilerhides much of the type-complexity from the user by providing a range of “syntactic sugar.” Figure 3 illustratesLINQ’s syntax with a fragment of a simple example program that computes the top-ranked results for each queryin a stored corpus. Two versions of the same LINQ expression are shown, one using a declarative SQL-likesyntax, and the second using the object-oriented style weadopt for more complex programs.The program first performs a Join to “look up” thestatic rank of each document contained in a scoreTriplestuple and then computes a new rank for that tuple, combining the query-dependent score with the static score inside the constructor for QueryScoreDocIDTriple. Theprogram next groups the resulting tuples by query, andoutputs the top-ranked results for each query. The fullexample program is included in [38].The second, object-oriented, version of the exampleillustrates LINQ’s use of C#’s lambda expressions. TheJoin method, for example, takes as arguments a datasetto perform the Join against (in this case staticRank) andthree functions. The first two functions describe how todetermine the keys that should be used in the Join. Thethird function describes the Join function itself. Note thatthe compiler performs static type inference to determinethe concrete types of var objects and anonymous lambdaexpressions so the programmer need not remember (oreven know) the type signatures of many subexpressionsor helper functions.3.2DryadLINQ ConstructsDryadLINQ preserves the LINQ programming modeland extends it to data-parallel programming by defininga small set of new operators and datatypes.The DryadLINQ data model is a distributed implementation of LINQ collections. Datasets may still con-4// SQL-style syntax to join two input sets:// scoreTriples and staticRankvar adjustedScoreTriples from d in scoreTriplesjoin r in staticRank on d.docID equals r.keyselect new QueryScoreDocIDTriple(d, r);var rankedQueries from s in adjustedScoreTriplesgroup s by s.query into gselect TakeTopQueryResults(g);// Object-oriented syntax for the above joinvar adjustedScoreTriples scoreTriples.Join(staticRank,d d.docID, r r.key,(d, r) new QueryScoreDocIDTriple(d, r));var groupedQueries adjustedScoreTriples.GroupBy(s s.query);var rankedQueries groupedQueries.Select(g TakeTopQueryResults(g));Figure 3: A program fragment illustrating two ways of expressing thesame operation. The first uses LINQ’s declarative syntax, and the second uses object-oriented interfaces. Statements such as r r.keyuse C#’s syntax for anonymous lambda expressions.NET objectsPartitionCollectionFigure 4: The DryadLINQ data model: strongly-typed collections of.NET objects partitioned on a set of computers.tain arbitrary .NET types, but each DryadLINQ dataset isin general distributed across the computers of a cluster,partitioned into disjoint pieces as shown in Figure 4. Thepartitioning strategies used—hash-partitioning, rangepartitioning, and round-robin—are familiar from parallel databases [18]. This dataset partitioning is managedtransparently by the system unless the programmer explicitly overrides the optimizer’s choices.The inputs and outputs of a DryadLINQ computationare represented by objects of type DryadTable T ,which is a subtype of IQueryable T . Subtypes ofDryadTable T support underlying storage providersthat include distributed filesystems, collections of NTFSfiles, and sets of SQL tables. DryadTable objects mayinclude metadata read from the file system describing table properties such as schemas for the data items contained in the table, and partitioning schemes which theDryadLINQ optimizer can use to generate more efficientexecutions. These optimizations, along with issues such8th USENIX Symposium on Operating Systems Design and ImplementationUSENIX Association

as data serialization and compression, are discussed inSection 4.The primary restriction imposed by the DryadLINQsystem to allow distributed execution is that all the functions called in DryadLINQ expressions must be sideeffect free. Shared objects can be referenced and readfreely and will be automatically serialized and distributedwhere necessary. However, if any shared object ismodified, the result of the computation is undefined.DryadLINQ does not currently check or enforce the absence of side-effects.The inputs and outputs of a DryadLINQ computation are specified using the GetTable T andToDryadTable T operators, e.g.:var input GetTable LineRecord ("file://in.tbl");var result MainProgram(input, .);var output ToDryadTable(result, "file://out.tbl");Tables are referenced by URI strings that indicate thestorage system to use as well as the name of the partitioned dataset. Variants of ToDryadTable can simultaneously invoke multiple expressions and generate multiple output DryadTables in a single distributed Dryadjob. This feature (also encountered in parallel databasessuch as Teradata) can be used to avoid recomputing ormaterializing common subexpressions.DryadLINQ offers two data re-partitioning operators:HashPartition T,K and RangePartition T,K .These operators are needed to enforce a partitioning onan output dataset and they may also be used to override the optimizer’s choice of execution plan. From aLINQ perspective, however, they are no-ops since theyjust reorganize a collection without changing its contents. The system allows the implementation of additional re-partitioning operators, but we have found thesetwo to be sufficient for a wide class of applications.The remaining new operators are Apply and Fork,which can be thought of as an “escape-hatch” that a programmer can use when a computation is needed that cannot be expressed using any of LINQ’s built-in operators. Apply takes a function f and passes to it an iterator over the entire input collection, allowing arbitrarystreaming computations. As a simple example, Applycan be used to perform “windowed” computations on asequence, where the ith entry of the output sequence isa function on the range of input values [i, i d] for afixed window of length d. The applications of Apply aremuch more general than this and we discuss them further in Section 7. The Fork operator is very similar toApply except that it takes a single input and generatesmultiple output datasets. This is useful as a performanceoptimization to eliminate common subcomputations, e.g.to implement a document parser that outputs both plaintext and a bibliographic entry to separate tables.USENIX AssociationIf the DryadLINQ system has no further informationabout f, Apply (or Fork) will cause all of the computation to be serialized onto a single computer. Moreoften, however, the user supplies annotations on f thatindicate conditions under which Apply can be parallelized. The details are too complex to be described inthe space available, but quite general “conditional homomorphism” is supported—this means that the applicationcan specify conditions on the partitioning of a datasetunder which Apply can be run independently on eachpartition. DryadLINQ will automatically re-partition thedata to match the conditions if necessary.DryadLINQ allows programmers to specify annotations of various kinds. These provide manual hints toguide optimizations that the system is unable to performautomatically, while preserving the semantics of the program. As mentioned above, the Apply operator makesuse of annotations, supplied as simple .NET attributes, toindicate opportunities for parallelization. There are alsoResource annotations to discriminate functions that require constant storage from those whose storage growsalong with the input collection size—these are used bythe optimizer to determine buffering strategies and decide when to pipeline operators in the same process. Theprogrammer may also declare that a dataset has a particular partitioning scheme if the file system does not storesufficient metadata to determine this automatically.The DryadLINQ optimizer produces good automaticexecution plans for most programs composed of standardLINQ operators, and annotations are seldom needed unless an application uses complex Apply statements.3.3Building on DryadLINQMany programs can be directly written using theDryadLINQ primitives. Nevertheless, we have begun tobuild libraries of common subroutines for various application domains. The ease of defining and maintainingsuch libraries using C#’s functions and interfaces highlights the advantages of embedding data-parallel constructs within a high-level language.The MapReduce programming model from [15] canbe compactly stated as follows (eliding the precise typesignatures for clarity):public static MapReduce( // returns set of Rssource, // set of Tsmapper, // function from T MskeySelector, // function from M Kreducer // function from (K,Ms) Rs){var mapped source.SelectMany(mapper);var groups mapped.GroupBy(keySelector);return groups.SelectMany(reducer);}8th USENIX Symposium on Operating Systems Design and Implementation5

Section 4 discusses the execution plan that is automatically generated for such a computation by theDryadLINQ optimizer.We built a general-purpose library for manipulatingnumerical data to use as a platform for implementingmachine-learning algorithms, some of which are described in Section 5. The applications are written astraditional programs calling into library functions, andmake no explicit reference to the distributed nature ofthe computation. Several of these algorithms need to iterate over a data transformation until convergence. In atraditional database this would require support for recursive expressions, which are tricky to implement [14]; inDryadLINQ it is trivial to use a C# loop to express theiteration. The companion technical report [38] containsannotated source for some of these algorithms.4System ImplementationThis section describes the DryadLINQ parallelizingcompiler. We focus on the generation, optimization, andexecution of the distributed execution plan, corresponding to step 3 in Figure 2. The DryadLINQ optimizer issimilar in many respects to classical database optimizers [25]. It has a static component, which generates anexecution plan, and a dynamic component, which usesDryad policy plug-ins to optimize the graph at run time.4.1Execution Plan GraphWhen it receives control, DryadLINQ starts by converting the raw LINQ expression into an execution plangraph (EPG), where each node is an operator and edgesrepresent its inputs and outputs. The EPG is closely related to a traditional database query plan, but we usethe more general terminology of execution plan to encompass computations that are not easily formulated as“queries.” The EPG is a directed acyclic graph—theexistence of common subexpressions and operators likeFork means that EPGs cannot always be described bytrees. DryadLINQ then applies term-rewriting optimizations on the EPG. The EPG is a “skeleton” of the Dryaddata-flow graph that will be executed, and each EPGnode is replicated at run time to generate a Dryad “stage”(a collection of vertices running the same computationon different partitions of a dataset). The optimizer annotates the EPG with metadata properties. For edges,these include the .NET type of the data and the compression scheme, if any, used after serialization. For nodes,they include details of the partitioning scheme used, andordering information within each partition. The outputof a node, for example, might be a dataset that is hashpartitioned by a particular key, and sorted according tothat key within each partition; this information can be6used by subsequent OrderBy nodes to choose an appropriate distributed sort algorithm as described below inSection 4.2.3. The properties are seeded from the LINQexpression tree and the input and output tables’ metadata,and propagated and updated during EPG rewriting.Propagating these properties is substantially harder inthe context of DryadLINQ than for a traditional database.The difficulties stem from the much richer data modeland expression language. Consider one of the simplestoperations: input.Select(x f(x)). If f is a simpleexpression, e.g. x.name, then it is straightforward forDryadLINQ to determine which properties can be propagated. However, for arbitrary f it is in general impossible to determine whether this transformation preservesthe partitioning properties of the input.Fortunately, DryadLINQ can usually infer propertiesin the programs typical users write. Partition and sort keyproperties are stored as expressions, and it is often feasible to compare these for equality using a combinationof static typing, static analysis, and reflection. The system also provides a simple mechanism that allows usersto assert properties of an expression when they cannot bedetermined automatically.4.2DryadLINQ OptimizationsDryadLINQ performs both static and dynamic optimizations. The static optimizations are currently greedyheuristics, although in the future we may implementcost-based optimizations as used in traditional databases.The dynamic optimizations are applied during Dryad jobexecution, and consist in rewriting the job graph depending on run-time data statistics. Our optimizations aresound in that a failure to compute properties simply results in an inefficient, though correct, execution plan.4.2.1Static OptimizationsDryadLINQ’s static optimizations are conditional graphrewriting rules triggered by a predicate on EPG nodeproperties. Most of the static optimizations are focusedon minimizing disk and network I/O. The most importantare:Pipelining: Multiple operators may be executed in asingle process. The pipelined processes are themselvesLINQ expressions and can be executed by an existingsingle-computer LINQ implementation.Removing redundancy: DryadLINQ removes unnecessary hash- or range-partitioning steps.Eager Aggregation: Since re-partitioning datasets isexpensive, down-stream aggregations are moved infront of partitioning operators where possible.8th USENIX Symposium on Operating Systems Design and ImplementationUSENIX Association

4.2.3Optimizations for OrderByDryadLINQ’s logic for sorting a dataset d illustratesmany of the static and dynamic optimizations availableto the system. Different strategies are adopted dependingon d’s initial partitioning and ordering. Figure 5 showsthe evolution of an OrderBy node O in the most complex case, where d is not already range-partitioned bythe correct sort key, nor are its partitions individually ordered by the key. First, the dataset must be re-partitioned.The DS stage performs deterministic sampling of the input dataset. The samples are aggregated by a histogramvertex H, which determines the partition keys as a function of data distribution (load-balancing the computationin the next stage). The D vertices perform the actual repartitioning, based on the key ranges computed by H.Next, a merge node M interleaves the inputs, and a Snode sorts them. M and S are pipelined in a single process, and communicate using iterators. The number ofpartitions in the DS H D stage is chosen at run timeUSENIX AssociationDSHODSHD(1)DSHDDD(2)Dynamic OptimizationsD(3)MMMMMSSSSSFigure 5: Distributed sort optimization described in Section 4.2.3.Transformation (1) is static, while (2) and (3) are Q makes use of hooks in the Dryad API todynamically mutate the execution graph as informationfrom the running job becomes available. Aggregationgives a major opportunity for I/O reduction since it canbe optimized into a tree according to locality, aggregating data first at the computer level, next at the rack level,and finally at the cluster level. The topology of such anaggregation tree can only be computed at run time, sinceit is dependent on the dynamic scheduling decisionswhich allocate

Figure 2: LINQ-expression execution in DryadLINQ. Step 1. A .NET user application runs. It creates a DryadLINQ expression object. Because of LINQ's ex-ecution of the expression has not occurred. Step 2. The application calls ToDryadTable trigger-ing a data-parallel execution. The expression .