{'fraction_non_alphanumeric': 0.030278321374211783, 'fraction_numerical': 0.008190186272378053, 'mean_word_length': 5.079422780348095, 'pattern_counts': {'":': 0, '<': 0, '<?xml version=': 0, '>': 1, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
SOFA: An Extensible Logical Optimizer for UDF-heavy Dataflows

Recent years have seen an increased interest in large-scale analytical dataflows on non-relational data. These dataflows are compiled into execution graphs scheduled on large compute clusters. In many novel application areas the predominant building blocks of such dataflows are user-defined predicates or functions (UDFs). However, the heavy use of UDFs is not well taken into account for dataflow optimization in current systems. SOFA is a novel and extensible optimizer for UDF-heavy dataflows. It builds on a concise set of properties for describing the semantics of Map/Reduce-style UDFs and a small set of rewrite rules, which use these properties to find a much larger number of semantically equivalent plan rewrites than possible with traditional techniques. A salient feature of our approach is extensibility: We arrange user-defined operators and their properties into a subsumption hierarchy, which considerably eases integration and optimization of new operators. We evaluate SOFA on a selection of UDF-heavy dataflows from different domains and compare its performance to three other algorithms for dataflow optimization. Our experiments reveal that SOFA finds efficient plans, outperforming the best plans found by its competitors by a factor of up to 6.


Introduction
In recent years, the characteristics of data analysis tasks have changed significantly. One change is the increase in the typical amounts of data to be processed; in addition, the diversity of the data to be analyzed and the heterogeneity of analysis tasks has grown considerably. While in the past most analytics were performed on structured data using relational data processors, such as SQL OLAP, many applications today require complex analyses, such as heavy-weight machine learning, predictive modeling, or graph traversal on large texts, graphs, semi-structured datasets, etc. Data processing systems support such analyses by providing system APIs for user-defined functions (Udf). Commonly, these Udfs are specified with imperative code, registered with the system, and called during execution. In fact, a large portion of Map/Reduce's popularity can be accounted to its widespread support for custom data processing [7].
A variety of dataflow languages has been proposed that aim at (a) making the definition of complex analytics tasks easier and at (b) allowing flexible deployment of such dataflows on diverse hardware infrastructures, especially compute clusters or compute clouds [24]. Many of these languages support Udfs [2,14,20]. Research has shown that proper optimization of such dataflows can improve the execution times by orders of magnitude [3,16,27]. However, most of these systems focus on relational operators and treat Udfs essentially as black boxes, considerably hampering optimization. At the same time, non-standard applications in areas such as information extraction, graph analysis, or predictive modeling often utilize Udf-rich dataflows. Traditional optimizers focus on relational operations, because their semantics in terms of optimization are well understood. In contrast, Map/Reduce-style Udfs can exhibit all sorts of behavior, which are difficult to describe in an abstract, optimizerenabling manner.
Different approaches have been proposed to overcome this problem. Two broad classes can be discerned: Approaches that require manual annotations of Udfs [1,17] Figure 1: High-level dataflow for employee relationship analysis. (a) Initial dataflow, (b) reordered dataflow based on operator semantics. and approaches performing code analysis to automatically infer optimization options with Udfs [3,11,16]. We present Sofa, a semantics-aware optimizer for Udf-heavy dataflows 1 . Compared to previous work, Sofa features a richer, yet concise set of properties for automatic and manual Udf annotation. It is capable of finding a much larger and more efficient set of semantically equivalent execution plans for a given dataflow than other systems. Given a concrete dataflow, both automatically detected and manually created annotations are evaluated by a costbased optimizer, which uses a concise set of rewrite templates to infer semantically equivalent execution plans.
Example. We use the following running example to explain the principles of Sofa throughout this paper: A large set of news articles shall be analyzed to identify persons, companies, and associations of persons to companies. We assume the articles stem from a web crawl and have already been stripped of HTML tags, advertisements, etc.; still the set contains many duplicate articles, as different news articles are often copied from reports prepared by a news agency.
An exemplary dataflow for this task is shown in Figure 1(a). The first operator performs duplicate removal by first computing a grouping key followed by an analysis of each group for similar documents, such that detected duplicates are filtered out per group. Next, a series of operators performs linguistic analysis (sentence splitting and part-of-speech tagging), entity recognition (persons and companies), and relation identification (persons ↔ companies). After each annotation operator, filter operators remove texts with no person, no company, or no person-company relation, respectively. As displayed in Figure 1(a), the dataflow is composed of nine steps: seven maps, and two complex operators (first and second from left).
If Udfs are treated as black boxes, this dataflow cannot be reordered in any way. But when provisioned with 1 Sofa is only a vague acronym but more of a metaphor.
proper information, such as data dependencies or operator commutativities, an optimizer has multiple options for reordering. For example, the part-of-speech tagger can be pushed multiple steps toward the end of the dataflow, as annotations produced by this operator are necessary for relationship annotation only. Moreover, the company and person annotation operators are commutative, as they independently add annotations to the text, but never delete existing annotations. Thus, both annotation operators can be reordered for early filtering. Figure 1(b) displays an equivalent dataflow with prospectively smaller costs as the most selective filters are executed as early as possible and expensive predicates are moved as much to the end of the dataflow as possible. As we see in Sections 3 and 7, existing dataflow optimization techniques, such as [16,20], cannot infer this plan.
A major obstacle to the optimization of Map/Reducestyle Udfs is their diversity. Our algorithm is developed in Stratosphere, a platform for data analytics on massive datasets with custom, domain-specific operator packages [1]. Available packages for information extraction (IE) and data cleansing (DC) already contain 38 and 9 operators, respectively. Further packages, e.g., for machine-learning and web analytics, are in development. Defining semantic properties for each of these operators individually would result in an unacceptable burden to the designer and would considerably limit extensibility and maintainability. Furthermore, the optimizer needs rewrite rules for operator pairs that take operator semantics into account. Thus, every new operator in principle has to be analyzed with respect to every existing operator to specify possible rewritings. Sofa solves this problem by means of an extensible taxonomy of operators, operator properties, and rewrite templates called Presto. Sofa uses the information encoded in Presto to reason about relationships between operators during plan optimization; specifically, it leverages subsumption relationships between operators to derive reorderings not explicitly modeled. Presto considerably eases extensions, as novel operators can be hooked to the system by specifying a single subsumption relationship to an existing operator exhibiting the same behavior with respect to optimization; these new operators are immediately optimized in the same manner as their parent. If desired and appropriate, more rewrite rules describing specific properties of the new operator may be introduced later in a "pay-asyou-go" manner [22].
In summary, our work includes the following contributions: 1. We identify a small yet powerful set of common Udf properties influencing important aspects in terms of dataflow optimization. 2. We show how properties of Udfs in Map/Reducestyle systems can be arranged in a concise taxonomy to simplify Udf annotation, and to enhance extensibility of dataflow languages. 3. We present a novel optimization algorithm that is capable of rewriting DAG-shaped dataflows given proper operator annotations. 4. We evaluate our approach using a diverse set of dataflows cutting through different domains. We show that Sofa subsumes three existing dataflow optimizers by enumerating a larger plan space and by finding more efficient plans. 5. Sofa outperforms the best plans found with other approaches in many situations with factors of up to 6.
This paper is structured as follows: Section 2 describes preliminaries including a brief overview of Stratosphere with a focus on its rich set of domain-specific Udfs. Section 3 gives an overview of our approach for dataflow optimization. Details on Presto and Sofa are explained in Sections 4 and 5. Section 6 discusses related work. We evaluate our approach in Section 7 and conclude in Section 8.

Preliminaries
We study the optimization of deterministic dataflows on a semi-structured data model, such as JSON or XML. A record is a potentially nested value tree consisting of objects, arrays, and atomic values. An unordered bag of records is called dataset. An operator o transforms a list of input datasets I = (I1, . . . , In) into a list of output datasets O = (O1, . . . , Om) by applying a Udf f to I. We denote with oin,i and oout,i the i-th input and output of operator o. We call S(oin,i) input schema and S(oout,i) output schema of the i-th input and output of o, respectively.
A dataflow is a connected directed acyclic graph D = (V, E) with the following properties: Vertices in D are either operators, data sources, or data sinks. Sources have no incoming and sinks no outgoing edges, respectively. Inner nodes with both incoming and outgoing edges are concrete operators. For any directed edge (u, v) connecting two operators we demand S(uout,j) ⊇ S(vin,i), and S(vin,i) must meet the schema requirements of the i-th input of v; according conditions hold for edges from an input dataset or to a data sink. Our data model does not require a schema definition in the first place, but operators might require that the processed records have a certain schema.
We call two deterministic dataflows D, D semantically equivalent (denoted D ≡ D ), if D and D produce the same output sets T , given the same input datasets I.

Stratosphere
The Sofa optimizer is integrated into Stratosphere, a full-fledged system for massively parallel data analytics. Meteor [14], a dataflow-oriented declarative scripting language resides at the top of the stack (see Figure 2). A Meteor query is parsed into an abstract syntax tree composed of basic or complex operators (see Section 3) and then compiled into a logical execution plan in the system's algebra called Sopremo. Sofa resides on the algebraic layer of Stratosphere and employs information on properties and semantics of operators stored in the Presto taxonomy (see Section 4) as well as statistics on the operators to perform dataflow optimization (see Section 5).
Meteor and Sopremo are designed for extensibility. Operators are defined in domain-specific packages, which are self-contained libraries of the operator implementations, their syntax, and semantic annotations. Algebraic Sopremo plans are compiled into Pact programs, which may consist of different parallelization primitives, such as map or reduce, and the associated user-defined function [1]. Subsequently, Pact programs are physically optimized and translated into an execution graph, which is deployed on the given hardware by means of Nephele, a system for scheduling, executing, and monitoring DAG structured execution graphs on distributed systems. A detailed description of Stratosphere can be found in [1]. Here, we describe only the algebraic optimizer.

User-Defined Operators
Before we introduce our Udf annotation schema and dataflow optimizer, we first introduce a subset of the op- erators currently used in Stratosphere to highlight their diversity and resulting optimization challenges. Userdefined operators are organized into packages specific to a certain application domain. Operators can be either abstract or concrete; for instance, anntt-ent-pers (see Table 1) is an abstract operator for annotating person names in texts, and its instantiations are different concrete algorithms and tools for performing this task. All instantiations of a given abstract operator share the same language syntax. Note that concrete operators may use very different implementations; for instance, the recognition of person names may be performed by using dictionary-, pattern-, or machine-learning-based methods. Concrete operators can either be elementary or complex. Elementary operators are implemented using a single stub function, complex operators are composed of several elementary operators similar to macros in programming languages. Complex operators are of high practical relevance, as they provide a shortcut for adding multiple elementary operators to a query. They are also important for dataflow optimization, since a complex operator may exhibit other semantics than those of its elements (see below).
Stratosphere currently contains three packages, namely a base package, a package for IE, and a package for DC. Packages for machine-learning and web data analytics are under development.  [14] for details. contains 16 operators, the IE package 38 operators, and the DC package 9 operators. The base package mostly comprises typical relational operators, such as filter, projection, transformation, join, and group. These operators are complemented by operators for semi-structured data, such as nest or unnest.
The IE package comprises three classes of operators: One for producing text annotations, one to merge annotations, and one for complex operators. Operators analyze the text and add, remove, or update annotations to the record. They may also transform records, e.g., the splt-sent operator takes as input single records formed of documents and outputs a set of records formed of sentences. The most abstract operator in the annotation class is anntt. Specializations can be distinguished between those performing linguistic annotations, semantic annotation of entities, or semantic annotations of relationships between entities. Each of these classes consists of multiple concrete operators; e.g., operators for tokenization, or part-of-speech tagging (first group), for recognizing persons, companies, or biomedical entities (second group), and for detecting binary or n-ary relationships between entities (third group). Specializations of anntt write to designated attributes in the output record; for instance, all entity operators write to a list-valued field "entities". Some anntt specializations are in precedence relations with other anntt variants, for example, annotating relations between entities requires that entity annotations are already present in the input records. The merge operator mrg merges records a, b from two input sets A, B based on a user-defined merge condition. The set of complex operators comprises six operators, such as operators for splitting text into sentences, stemming, entity extraction, and stopword removal. Each of these internally consists of an anntt operator, a trnsf operator, and occasionally a fltr operator.
The DC package comprises six different classes of operators for data cleansing and data integration [13] They address common challenges of dirty or heterogeneous data sources, such as inconsistent representation of equivalent values, fuzzy duplicates, typographic errors, or missing values. Inconsistencies and missing values can be fixed with the scrb operator that either repairs these values or filters invalid records. Fuzzy duplicates are found with ddup and lnkrc within one relation or across several relations, respectively. These duplicates can subsequently be coalesced into a single record with fuse. The complex duplicate removal operator (rdup) combines duplicate detection and fusion of duplicates to solve the common task of removing all duplicate records within one data source.
The algebraic plan for the dataflow for our running example is shown in Figure 3(a) together with properties and schema information (cf. Figure 3(b)), which are used for optimization. Figure 3(c) displays that complex operators may exhibit different properties than its elementary components: the complex operator splt-sent has different read/write set annotations and different I/O ratios than its elementary components. In the following section, we demonstrate how this plan can be reordered substantially by exploiting information on operator semantics.

Dataflow Optimization
Sofa is an optimizer for rewriting Udf-heavy dataflows into semantically equivalent dataflows whose expected efficiency is higher, according to a cost model. Rewriting depends on a set of rewrite rules, each defining valid manipulations of dataflow sub-plans, such as a reordering of two filter operations [9]. The novelty of Sofa is its flexible and extensible treatment of Map/Reduce-style Udfs going far beyond the capabilities of existing approaches. In this section, we highlight the advantages of Sofa by means of our running example.
Starting from a dataflow D, dataflows semantically equivalent to D may be produced using different transformation techniques. Sofa is capable of introducing, removing, and reordering operators. Complex operators are optimized both as a whole and after expansion. In the following, we focus on reordering and treatment of complex operators, which are the most intricate and most effective optimization techniques.
Existing approaches for dataflow optimization enable reorderings by using either manually defined rewrite rules for relational operators [20] or by performing some kind of code analysis [3,16]. The approach of Hueske et al. [16] probably is the most general, as it automatically derives dataflow reorderings based on read/write set analysis of individual Udfs. In particular, the order of two subsequent tuple-at-a-time operators may safely be switched if they have no read/write or write/write conflicts on any attribute. The dataflow shown in Figure 3 allows only one such reordering: The anntt-pos operator that annotates part-of-speech tags and stores them in the fourth attribute (first row, third from right) can be pushed before the anntt-rel operator (second row, second from right), because part-of-speech annotations are accessed only during relation annotation. This reordering most likely saves time, because the different fltr operators are now executed before anntt-pos and thus fewer sentences have to be annotated. Moving anntt-pos towards the start of the dataflow is not possible, because it reads annotations produced by the complex splt-sent operator.
Semantics-aware rewrite rules allow to reorder the dataflow in Figure 3 more extensively. Consider the two annttent operators. Both write into the same attribute (the fifth), which collects all entity information. If the optimizer knows that annotation operators only add values and never delete or update existing values, these operators together with their subsequent filter operators may be reordered. The best order very likely is the one that filters the most sentences first; this decision can make use of selectivity and execution time estimates at the operator level (see Section 5.3). Furthermore, the optimizer can decompose complex operators and reorder the components individually. For instance, splt-sent consists of an anntt-sent operator and a Udf splitting the input text into separate sentences based on the annotations produced by anntt-sent. As shown in Figure 3(c), the two components of splt-sent differ in terms of read and write access on attributes and I/O ratio. Pushing split-Udf some steps towards the end of the plan is valid, because all succeeding anntt operators perform their analyses sentence-based, since all anntt operators for entity, relation, and part-of-speech annotation read sentence annotations. This reordering is likely beneficial, because the split-Udf generates multiple output records for each   incoming record, depending on the number of annotated sentence boundaries. Note that sometimes the expansion of complex operators may make possible reorderings impossible. For instance, think of an IE operator anntt-syns for expanding existing entity annotations with all synonyms from a dictionary. Subsequently, another operator repl-repr selects one of these synonyms as representative and deletes all others. Both operators change the existing list of entity annotations, so reordering based on read/write-set analysis is limited. However, a complex operator norment for normalizing entity annotations composed of both anntt-syns and repl-repr is optimizable, since we know that the number of entities does not change. Therefore, Sofa always tries to reorder complex operators both as a whole and after their expansion.
In summary, semantics-aware plan rewriting allows us to pick the best plan (with respect to cost estimates) from a larger set of equivalent dataflows compared to other existing approaches. For instance, Sofa finds 4,545 distinct plans for the running example, compared to only 512 plans found with the read/write-set analysis of [16] (see Section 7 for a detailed comparison).

Annotating and Rewriting Operators with Presto
To enable optimizations like those shown in the previous section, operators need to be annotated with meta data, for instance to describe selectivity estimates or semantic properties, such as associativities or commutativities. In this section, we introduce Presto, an extensible taxonomy for annotating and rewriting operators. Presto consists of two major components, the operator-property graph for modeling relationships between operators and properties (see Section 4.1), and a set of rewrite templates for dataflow-rewriting (see Section 4.2). When designing Presto, we paid special attention to extensibility by allowing enhancements to the semantic operator descriptions over time, to more and more unleash their optimization potential (see Section 4.3).

Operator-Property Graph
The operator-property graph in Presto contains two taxonomies for classifying operator properties. Both taxonomies are self-contained and model generalization-specialization relationships (isA) between operators and properties, respectively. become much deeper when dealing with domain-specific Udfs.
As shown in Figure 4(b), we distinguish between automatically detectable properties and properties that are annotated by the package developer. The latter comprises algebraic properties (e.g., commutativity, associativity, or idempotence), cost model (e.g., cost functions, resource consumption), and the ratio between the number of input and output records. Automatically detectable properties comprise the parallelization function of the operator implementation (e.g., map, reduce), schema information available at query compile time, and the read/write behavior at attribute level.
Relationships connect operators and properties. As usual, each specialization inherits all properties and relationships that are defined for the corresponding generalizations. For instance, the union-all shown in Figure 4(c) is a specialization of the union operator and thus inherits the algebraic properties defined for union. Complex operators can be characterized with respect to their components using the hasPart relation (Figure 4(d)). For example, the complex operator splt-sent consists of the two components anntt-sent and split-UDF. Next to isA and hasPart, we define a hasProperty and a hasPrerequisite relation.
HasProperty is a binary relation between an operator and a property and is used to characterize operator semantics. For instance, the following properties are attached to fltr (Figure 4(c)): -does not modify the schema of incoming records, -is implemented with a Map function, -does not modify inside fields, -input ≥ output, -processes one record at a time, and -is commutative to other fltr instantiations.
Precedence constraints between operators are captured with hasPrerequisite(X,Y), which states that operator X must be executed before operator Y . In Figure 4(d) it is shown that anntt-rel based on linguistic patterns requires part-of-speech and entity annotations to be performed in advance. Since anntt-ent itself requires sentence annotation and hasPrerequisite is a transitive relation, it is necessary to apply anntt-sent before anntt-rel.
The isA relationship very much simplifies deriving novel rewrite options for operators that are initially not well annotated. Suppose, the data scrubbing operator scrb from the DC package is initially not equipped with any hasProperty relationships. Later, the developer may see that scrb is a specialization of the well-annotated Base operator trnsf, i.e., both operators perform write operations in attributes of the incoming records. By formally specifying this through an isA relationship, scrb inherits all properties defined for trnsf (not shown in Figure 4(a)).
Though the complete Presto graph is too large to show here, it is still rather small and easy to understand: The property taxonomy contains 32 nodes and the operator taxonomy 78 nodes. Note that new packages mostly ex-tend the operator taxonomy, while the property taxonomy is a fairly stable structure in our experience.

Rewrite Templates
We perform dataflow-rewriting using a set of rules specifying semantically valid reorderings, insertions, or deletions of operators. Because rewrite rules apply to combinations of operators, and because the different independently developed and maintained packages available for Stratosphere already contain more than 60 individual operators, it is practically impossible to define all rewrite rules across the different packages one-by-one. Instead, we define a concise set of rewrite templates using Presto relationships and abstract operators as building blocks. Reasoning along relationships allows Sofa to automatically instantiate the templates with concrete operators and thus enables us to derive individual rewrite options for concrete operator combinations on-the-fly. Currently, Sofa requires only 10 rewrite templates, which are expanded to over 150 individual rewrite rules. Figure 5 displays a subset of the available templates in Datalog notation; further rules cover different reorderings based on algebraic properties as well as insertion and removal of operators (not shown here for brevity). The first three templates are static and can be evaluated at package loading time, whereas the last two templates are dynamic and are evaluable at query compile time only. The first template covers commutative operators and expresses that two consecutive appearances of operators X annotated as commutative in Presto can be safely reordered. Specifically, the goal reorder(X,X) evaluates to true if Presto contains a hasProperty-relationship of X with the property commutative. Note that the commutativity does not necessarily need to be defined directly on X; the rule also applies if some ancestor of X in Presto is marked as commutative. The second template (Line 4) enables reordering of operators based on the isA relation and states that for any three operator instantiations X, Y, Z, the operators X, Y are reorderable given that X is not a prerequisite of Y , X is a specialization of Z, and Y, Z are reorderable. We include the goal not hasPrerequisite(Y,X) in the templates to ensure that operator precedences are respected. The third template (Lines 6-7) enables reorderings of consecutive anntt operators X, Y , when X is not a prerequisite of Y .
Dynamic rewrite templates are partly based on information not available before the query is compiled, for example, information on concrete attribute access by operators is only available after posing a query. Tem- plate 4 (Lines 10-12) enables reordering of two singleinput record-at-a-time operators if these operators have no read/write conflicts. This single rule essentially covers most optimization options achieved by [16], which shows the power of our approach.
While most rules in Presto are generic and apply to many operator combinations, other rules are more specific. Suppose, we are given a dataflow that consists of an equi-join of two data sources I1, I2 followed by trnsf that transforms only attributes of I1, which are not part of the join condition. This dataflow can be rewritten into an equivalent dataflow, which first applies trnsf to I1 and afterwards joins I1 and I2: Similar to extending Presto with new operators and properties, package developers can also extend the set of rewrite templates to enable dataflow optimization for their concrete application domain. For example, the third template was added by the IE package developer, since it enables reordering anntt instantiations, which is not supported by any other Presto template.

Pay-as-you-go annotation of operators
A key feature of Sofa is its extensible design. Consider a new package for web analytics to be integrated into Stratosphere, containing an operator rmark for detecting and removing HTML markup in web pages. Initially, this operator would probably not be equipped with any Presto annotations. In this case, the Sofa optimizer can infer only automatically detectable properties, i.e., reordering can only be performed on the basis of read/write-set analysis. Later, the package developer invests some thought and annotates that rmark outputs as many records as incoming (|I| = |O|). Sofa infers from the set of automatically detectable properties, that rmark is a single-input operator implemented with a map function and does not change the schema of the incoming records. Taken these properties together, the last template of Figure 5 becomes applicable to rmark. A full specification of rmark would include the definition of isA relationships to other operators. Actually, rmark has the same semantics as the trnsf operator from the Base package, as it essentially performs a transformation of the input texts. Now all templates valid for trnsf become applicable, such as the rule for reordering a join and a trnsf operator introduced in Section 4.2. Given that rmark accesses only attributes present in input I1 that are not part of the join condition, Sofa can then reorder a dataflow containing rmark and join as follows:

Algorithms
Given a dataflow D, the Sofa optimizer performs two passes of the following three steps. First, D is analyzed for precedence relationships between operators based on rewrite templates and operator properties contained in Presto. This analysis yields a precedence graph, which is used in the plan enumeration phase, to secondly enumerate and thirdly rank valid plan alternatives based on a cost model. Afterwards, the complex operators contained in D are resolved into their elementary components and the three steps are repeated. Finally, the best plan is selected, translated, and physically optimized for parallel execution by the underlying execution engine (see [1] for details on this step). In the following, we discuss Phases 2-4 in more detail.

Precedence analysis
Presto models dependencies between operators either explicitly on the basis of the hasPrecedence relation or inferred, if the goal reorder(X,Y) fails for two operator instantiations X, Y . The precedence graph construction starts by creating the directed transitive closure of the given dataflow, which explicitly models all pairwise operator execution orders in D. The algorithm then in- spects this graph to detect and remove edges that are not logically required. It retains all edges incident to a data source or a sink to prevent reordering of sources and sinks. The goal reorder(X,Y) is instantiated with start and end node of each edge (u, v) and the inference mechanism tries to resolve the goal based on the operator properties and rewrite templates stored in Presto. If successful, both nodes are reorderable and the edge (u, v) is removed from the precedence graph. Precedence analysis is a polynomial time algorithm; its complexity is determined by computing the transitive closure in O(|V | 3 ) using the Floyd-Warshall algorithm and the data complexity of stratified non-recursive Datalog, which we use for reasoning in Presto [6]. Figure 6 shows the final precedence graph for our running example (omitting data sources and sinks for readability). The displayed graph reflects precedences between DC, IE, and Base operators, e.g., rdup and annttent-person are a prerequisite for the fltrperson>0 operator, and anntt-rel is in a hasPrerequisite relation with annttpos (cf. Figure 4(d)). The graph contains edges between anntt and fltr reflecting that the concrete instantiations of fltr have read/write conflicts with their preceding anntt operators.

Plan enumeration
Plan enumeration essentially generates different topological orders given by the precedence graph, while performing cost-based pruning. On contrast to topological sorting, the outcome are not full orders but DAG-shaped plans. The main idea is to construct alternative plans for a given dataflow D iteratively by analyzing the corresponding precedence graph for operators that have no outgoing edges. Such operators are not required by any other operator and can therefore be added to the emerging physical dataflows. If multiple operators have no outgoing edges, the algorithm creates a set of alternative partial plans. The algorithm continues to pursue each alternative, removing the newly added operator from the precedence graph, estimating the costs of the partial plan (see Section 5.3), and pruning costly partial plan alternatives where possible.
We explain its principles using the simplified dataflow shown in Figure 7 (top). Note that this dataflow is DAGshaped, which poses no problem to Sofa. The dataflow performs task-parallel annotation of persons and companies. Annotations are subsequently merged, and the result set is filtered for articles published after 2010. The resulting precedence graph is displayed in Figure 7 (bottom).
The recursive plan enumeration algorithm is displayed in Figure 8. It takes as input the original dataflow, the corresponding precedence graph, and a partial plan, which initially is empty (Lines 1-2). First, the algorithm selects the set of nodes from the precedence graph that have out-degree 0 (Line 8). These operators are not a prerequisite of any remaining operator and can thus be added to the partial plan (Lines 11-13) without violating precedence constraints. In our example, only the data sink can be selected. Once added, the selected node is removed from the precedence graph (Line 14). Since the partial plan was empty before adding the data sink, we cannot insert any edges in the partial plan and therefore, plan enumeration is recursively invoked again (Lines [16][17]. Now, mrg and fltr both have no outgoing edges anymore and are therefore added to the set of candidate nodes. Each candidate node is processed individually, added to the partial plan and removed from the precedence graph. This yields in two alternative partial plans, which are both inspected further. We exemplarily follow the plan with the mrg operator. The operator is added to the plan and subsequently, the set of inputN odes is divided into required and optional nodes (Lines [19][20][21][22][23][24]. Required nodes are those nodes that have the currently added node as its direct predecessor 1 enumAlternatives(Graph precedGraph, Graph plan, 2 Graph partialPlan) { 3 4 if (isEmpty(precedGraph)) { 5 addPlanToResultSet(partialPlan); 6 return; 7 } 8 candidateNodes = getNodesWithOutDegreeZero(precedGraph); 9 10 foreach(Node n in candidateNodes) { 11 inputNodes = getNodesWithOpenInputs(partialPlan); 12 13 addNodeToPartialPlan(n); 14 removeNodeAndIncidentEdgesFromPrecedenceGraph(n); 15 16 if (isEmpty(inputNodes)) 17 enumAlternatives(precedGraph, plan, partialPlan); 18 19 foreach(Node m in inputNodes){ in the original dataflow, optional successors are all other operators contained in inputN odes. In our example, the set of required nodes is empty, and the set of optional nodes contains fltr. For each required node m, we create an edge (n, m) for the newly added node n, add it to the edge set of our partial plan, estimate the costs of the partial plan, and recursively call the plan enumeration algorithm (Lines 25-29). Each optional node l is processed individually. We iteratively create edges (n, l), estimate the costs of the new partial plan, and again recursively call the plan enumeration algorithm if necessary (Lines 30-34). A recursive invocation of the plan enumeration algorithm terminates either if the precedence graph is empty and an alternative plan has been found (Lines 4-7), or if no alternative plans with smaller costs compared to the initial plan were found (Line 33). Figure 9 shows all stages of enumerating the plan space for the dataflow from Figure 7. In Stage 1, only the data sink can be selected (grey box) and is thus removed in Stage 2. The algorithm can now either add mrg or fltr (red and green boxes) as both have no outgoing edges in Stage 3. In Stage 4, the mrg plan results in three further alternatives, namely fltr, anntt-ent-comp, and anntt-entpers. At that point, only the source cannot be added to the plan yet and consequently the remaining two operators can be added in arbitrary order. Thus, the four alternatives are expanded to eight plans in Stage 5. Finally, the last operator and source is added in Stage 6 and 7 resulting in 12 different alternatives. Pruning. The plan enumeration algorithm has exponential worst-case complexity (consider for instance a precedence graph without any edges). We included a simple technique for search space pruning in our algorithm preventing completion of partial plans whose estimated costs are higher than the estimated costs for the current best dataflow. Once a cheaper plan was found, we update the costs of the best plan, in a manner similar to accumulated cost pruning in top-down query optimization [9,10]. If no alternative plan with lower estimated costs compared to the best plan could be constructed, we terminate (cf. Figure 8, Line 33).

Cost estimation
To estimate costs and result sizes of a dataflow, Sofa depends on estimates for key figures of operators, which can either be provided by the developer by adding appropriate annotations to Presto, by sampling from the input data, or by runtime monitoring of previously executed dataflows. We estimate the costs of a plan by computing the weighted sum of estimated ship data volume, I/O volume, and CPU usage of the Udfs per stub call.
Specifically, the costs of an operator oi are estimated as follows: let ci be the average CPU usage of oi per invocation, si the estimated startup costs of oi, and ri the estimated number of processed input items of oi. Including startup times of operators is particularly important for complex non-relational Udfs, as many IE and DC operators need a long startup time for instance to load large dictionaries, or to assemble trained models. Furthermore, let di denote the estimated I/O costs of an input item processed by oi, ni the estimated shipping costs of an output item produced by oi, and seli the selectivity of oi. The estimated number of items ri processed by an operator oi is calculated as ri = (h,i)∈E(D) r h * sel h . The costs of an operator oi are estimated as the weighted sum of the estimated ship data volume, I/O volume, and CPU usage of oi using the formula costs(oi) = w * (ci * ri + si) + u * (di * ri) + v * (ni * ri * seli), where u, v, w denote weight constants for each cost component. Note that the formula for operator costs can be replaced with custom cost functions, which are added to Presto by the package developers. For example, to accurately estimate the costs for dataflows containing IE operators, we also capture the projectivity of anntt operators, i.e., the average number of annotations produced by an anntt instantiation. Consequently, the selectivity of fltranntt is denoted as sel(f ltranntt) = ri−1 * proj(anntt).
Finally, the total costs of a dataflow D are estimated as costs(D) = n i=1 costs(oi). Note that our cost model optimizes for total computation time, disregarding parallelization in the underlying execution engine. However, we see in Section 7 that this approach already allows us to correctly rank enumerated plan alternatives in many cases.
rewrite query expressions and compile them to physical execution plans. Chaudhuri et al. proposed declarative rewrite rules to improve the optimization of queries with external functions [4]. All aforementioned approaches deal with the optimization of relational queries and assume that rewrite rules are manually added to the optimizer framework. Our work differs as it focuses on Udf-heavy query, also deals with DAG-shaped plans, and does not require the explicit definition of condition-action rules for new operators but automatically infers them from a concise set of operator properties. This considerably increases extensibility.
The optimization of relational queries with user-defined predicates has been another focus of research [5,15,26]. In these works, the semantics of the Udfs to be reordered is assumed to be uniform, i. e., filtering tuples. Consequently, these approaches focus on the problem of identifying the optimal order of filters and do not address the question whether general Udfs can be reordered or not; besides, they disregard the effects of parallelization functions on Udfs. In contrast, our work addresses the optimization of Map/Reduce-style dataflows with userdefined operators for which additional information is required to answer this question. In our approach, this information is provided by the developer or inferred from Presto.
While common query optimizers use tree-shaped query execution plans, our work focuses on DAG-shaped dataflows in the spirit of [18]. There, Neumann investigated the problem of generating and executing DAG-shaped query evaluation plans for relational queries in order to enable sharing of intermediate results. This is different from our setting as the input of our optimizer are already DAG-shaped dataflows in contrast to declarative relational queries.
Several methods to optimize more general dataflows have been proposed. Ogasawara et al. propose an algebraic approach to define scientific workflows [19] and optimize their execution. Operators are classified as Udfs with strict templates or as relational expressions. While their classification of Udfs is somewhat similar to particular combinations of our operator properties, we regard our approach as more general as it features a richer set of properties that can be freely composed to precisely reflect the characteristics of an operator. Furthermore, their work does not contain a transformation-based optimizer. Simitsis et al. present an approach for optimizing ETL processes [25]. They introduce three different optimization techniques, namely reorderings of adjacent single-input/single-output operators that have no read/write-conflicts, merging and splitting of operators, and factorization of operators. Our approach is more general as Sofa is able to optimize arbitrary dataflows and it is able to reorder any operator combinations given that precedence constraints are respected. Stubby is an optimizer for workflows constructed of multiple Map/Reduce jobs [17]. This approach is orthogonal to Sofa, as it leverages manual annotations of Map and Reduce functions to merge Map/Reduce jobs while preserving the logical order of operations. The MRQL framework performs physical algebraic optimizations of Map/Reduce dataflows consisting only of relational operators [8], but disregard general Udfs. Both approaches identify valid transformations from static rules and operator properties, such as in-and output schema. Our approach differs in using an extensible taxonomy of a richer set of operator properties and rewrite rules to deduce precedence constraints. Work presented by Hueske et al. [16] on reordering operators in Map/Reduce-style dataflows is based on specific information about the operator's behavior in terms of accessed data attributes obtained by static code analysis. We included this idea in Sofa, but also show how semantic annotations that describe the behavior of Udfs more precisely can help to further increase the space of possible rewritings; besides, our approach allows to rewrite DAG-shaped dataflows, which is not possible with the algorithm presented in [16]. The SUDO optimizer [28] combines manual annotation and code analysis to analyze Udf properties with respect to data partitioning to avoid unnecessary data shufflings. This problem is orthogonal to Sofa, where we analyze semantic operator properties to reduce execution times by reordering operators. Pig Latin [20] is a procedural higher-level language for Hadoop and features a "safe" optimizer. It applies a limited set of heuristic transformation rules, such as filter push down, that most likely is beneficial and relies otherwise on the decisions of the programmer.
In summary, we believe that Sofa is the first extensible, fully functional optimizer for arbitrary DAG-shaped dataflows for Map/Reduce-style systems.
plication domains 2 . Q1, Q2, and Q7 are pipeline-shaped, Q3 and Q6 are tree-shaped, and Q4 and Q5 are DAGshaped. Q1 adopts the dataflow described in our running example for relationship extraction from biomedical literature using IE and DC Udfs. Q2 performs an advanced word count by computing term frequencies in a corpus grouped by year. The query first splits the input data into sentences, reduces terms to their stem, removes stopwords, splits the text into tokens, and aggregates the token counts by year. Q3 extracts NASDAQ-listed companies that went bankrupt between 2010 and 2012 from a subset of Wikipedia. This query takes article versions from two different points in time, annotates company names in both sets and applies different fltr operators and a join to accomplish the task. Q4 corresponds to the dataflow shown in Figure 7 and performs task-parallel annotation of person and location names. Q5 analyzes DBpedia to retrieve politicians named 'Bush' and their corresponding parties using a mixture of DC and base operators. Q6 is a relational query inspired by the TPC-H query 15 . It filters the lineitem table for a time range,  joins it with the supplier table, groups the result by join key, and aggregates the total revenue to compute the final result. Q7 uses two complex IE operators to split incoming texts into sentences and to extract person names. Datasets. We evaluated Q1 on a set of 10 million randomly selected citations from Medline, Q2 was evaluated on a set of 100,000 full-text articles from the English Wikipedia initially published between 2008 and 2012, Q3 was evaluated on two sets of English Wikipedia articles of 50,000 articles each, one set from 2010 and one set from 2012, Q4 and Q7 on a set of 100,000 full-text articles from the English Wikipedia downloaded in 2012, Q5 on the full DBpedia dataset v. 3.8, and Q6 was evaluated on a 100GB relational dataset generated using the TPC-H data generator. For each experiment, we report the average of three runs. Estimates on operator selectivities, projectivities, startup costs, and average execution times per input item were derived from 5% random samples of each dataset. Competitors. Although dataflow languages for Big Data are a hot topic in current research, surprisingly few systems actually optimize the dataflow at the logical level as we do. Thus, detecting appropriate competitors is difficult, because optimizers are commonly deeply coupled to a particular system. We reimplemented the ideas of three current dataflow optimizers, namely techniques presented by Hueske et al. [16], Olston et al. [20], and Simitsis et al. [25]. We compare the number of plan alternatives found and the achieved runtime improvements. For each method, we disabled rules and information on operator properties stored in Presto and replaced them with the appropriate rewrite rules described in [16,20,25]. For the method of Olston et al., we referred to the online documentation of rewrite rules for Apache Pig, version 0.11.1. For Hueske et al., we enabled annotation of read-and write-sets, but disabled reordering of DAG-shaped plans.

Finding optimal plans
A large number of semantically equivalent plans for a concrete dataflow has the potential to contain the most effective variant. Therefore, we first evaluate Sofa to all three competitors with respect to the number of alternative plans found with each method. We turned search space pruning off and enumerated the complete space of alternative dataflows for all queries. In Section 4, we explained how complex operators can be resolved into a series of interconnected elementary operators. Q1, Q2, and Q7 contain complex operators, thus, we enumerated the plan space for these queries both using only elementary operators and using combinations of elementary and complex operators. For the methods presented in [16,20,25], we used complex operators only, as this methods do not provide mechanisms for operator expansion.
As displayed in Table 2, Sofa enumerates the largest plan space in all cases. The method presented by Hueske  To evaluate the correctness of plan ranking performed by Sofa, we sorted the complete plan space for each query plan ascending by estimated costs, selected different plans from a rank interval, and report estimated costs and observed runtimes for these plans. As shown in Figure 10, Sofa ranks the different algebraic execution plans correctly, and for Q1, Q2, Q5, and Q7, the best ranked plans were retrieved only with Sofa.
We also observed a large optimization potential for most tasks. For example, the best ranked plans for Q1-Q4 outperform the worst ranked plans with factors in the range of 4.2 (Q2) to 9.1 (Q1). For the remaining queries Q5-Q7 we observed differences in execution times of 23 to 28 % between the best and worst plan. Note that these three queries were the shortest running in our experiments with total runtimes between 10 to 30 minutes, and a significant portion of these runtimes can be attributed to system initialization and communication. Thus, we expect that these queries benefit much more from optimization on larger datasets. Table 2 displays the plan space with search space pruning enabled in brackets. For queries spanning the largest plan space (Q1 and Q3), pruning helps to significantly reduce the enumerated plan space. For the methods presented in [20,25], which both enumerate significantly smaller plan spaces than Sofa, pruning as performed by our enumeration algorithm does not reduce the plan space in most cases. For each tested query, the optimization time with pruning enabled takes not longer than 2.5 seconds with Sofa. Enumerating the complete plan space for each query takes at most 10 seconds, which is negligible compared to the execution times of our long-running evaluation queries. Note that the largest part of these optimization times can be attributed to reasoning along Presto relationships, which could be improved in many known ways [23].

Optimization benefits
In our third experiment, we evaluated to which extent dataflow optimization benefits from information on operator semantics. Figure 11 displays the execution times of the best ranked plan found with Sofa as well as the methods described in [16,20,25]. For each tested query, Sofa finds the fastest plan, and for Q1, Q2, Q5, and Q7, Sofa finds significantly faster plans than competitors: the best plan found with Sofa outperforms the best plans found by [16] with factors of up to 6.8 (Q4), by [20] and [25] with factors up to 4.2 (Q2). The method of Hueske et al. performs as well as Sofa for Q3 and Q6, because both methods enumerate the same plan space for these two queries. The rewrite rules of Olston et al. and Sim-itsis et al. find the same best plan as Sofa for Q4. In these cases, plan optimization involves only reordering filter operators, which is addressed equally well in these methods as in Sofa. Note that the method of Hueske et al. cannot rewrite Q4, as this query is DAG-shaped. All other queries involve rewriting general Udfs and expansion of complex operators, and thus, optimization benefits notably from semantic information that is available in Sofa.

Extensibility
Finally, we concretize the example from Section 4.3 to quantify the effect of pay-as-you-go annotation of operators in Sofa. Recall the novel rmark operator, which replaces HTML tags in web pages by a series of '%' of the same length as the removed tags to retain text length and markup position. Imagine a query Q8 that first replaces HTML markup in websites, computes term frequencies from the websites content, and finally filters terms starting with a series of '%'. The high-level dataflow looks as follows: O fltr grp splt-tok rm-stop stem splt-sent rmark I Initially, rmark is annotated only with an isArelationship to the abstract Presto concept operator. In this case, Sofa can analyze only read and write access on attributes similar to the method presented in [16], which yields in 10 semantically equivalent plans for Q8. After adding the information that rmark is a schema preserving record-at-a-time operator implemented with a Map function, Presto already finds 18 equivalent algebraic plans. Finally, when rmark is fully specified, including an isA relationship to the Base operator trnsf, Sofa would find 75 alternative plans.

Conclusions
We addressed the problem of logical optimization for Udf-heavy dataflows and present Sofa, a novel, extensible, and comprehensive optimizer. Sofa builds on a concise set of properties describing the semantics of Map/Reduce-style Udfs and a small set of rewrite templates to derive equivalent plans. A unique characteristic of our approach is extensibility: we arrange operators and their properties into a taxonomy, which considerably eases integration and optimization of new operators. We implemented our solution in Stratosphere, a fullyfunctional system for large-scale data analytics. Our experiments reveal that Sofa is able to reorder acyclic data-flows of arbitrary shape (pipeline, tree, DAG) from different application domains, leading to considerable runtime improvements. We also show that Sofa finds plans that clearly outperform those from other techniques.