PaperlayerView exact revision on arXiv

Source-linked AI summary

Opening the Black Boxes in Data Flow Optimization

Fabian Hueske, Mathias Peters, Matthias Sax, Astrid Rheinländer, Rico Bergmann, Aljoscha Krettek, Kostas Tzoumas

arXiv:1208.0087v1cs.DB

TL;DR

The paper asks how to optimize data flows containing imperative black-box UDFs when operator semantics are unknown. It uses a small set of statically inferred properties to establish reorderings and implements a corresponding optimizer. The approach reproduces common relational rewritings, improves relational and non-relational data flows, and supports optimizations unavailable to algebraic optimizers.

  • Problem

    Optimizing data flows with arbitrary imperative UDFs is difficult because their semantics are hidden, unlike the known algebraic semantics assumed by conventional query optimizers.

  • Method

    The optimizer formally derives reordering conditions from a few properties and estimates those properties by statically analyzing UDF code.

  • Results

    The system reproduces most relational reorderings, optimizes non-relational data flows, and achieves runtime improvements of up to an order of magnitude.

  • Takeaways & Limitations

    Black-box UDFs can be optimized across relational and non-relational data flows without prior knowledge of their full operator semantics.

  • Takeaways & Limitations

    The approach cannot reorder operations requiring semantic information such as associative side-effects, and exhaustive plan enumeration faces exponential search-space growth.

Abstract

from arXiv · show

Many systems for big data analytics employ a data flow abstraction to define parallel data processing tasks. In this setting, custom operations expressed as user-defined functions are very common. We address the problem of performing data flow optimization at this level of abstraction, where the semantics of operators are not known. Traditionally, query optimization is applied to queries with known algebraic semantics. In this work, we find that a handful of properties, rather than a full algebraic specification, suffice to establish reordering conditions for data processing operators. We show that these properties can be accurately estimated for black box operators by statically analyzing the general-purpose code of their user-defined functions. We design and implement an optimizer for parallel data flows that does not assume knowledge of semantics or algebraic properties of operators. Our evaluation confirms that the optimizer can apply common rewritings such as selection reordering, bushy join-order enumeration, and limited forms of aggregation push-down, hence yielding similar rewriting power as modern relational DBMS optimizers. Moreover, it can optimize the operator order of non-relational data flows, a unique feature among today's systems.

1. INTRODUCTION

The paper targets optimization of data flows built from imperative UDFs whose semantics are hidden from conventional optimizers. It derives reorderings from limited properties, extracts them through static analysis, and demonstrates relational and non-relational optimization.

  • Motivation: Imperative UDFs in data-flow systems hide operator semantics, making parallelization and operator reordering coupled optimization challenges.Traditional RDBMS optimizers generally support only restrictive UDF templates, whereas MapReduce-style UDFs use more general-purpose code.
  • Contributions: The paper introduces data-flow reordering for arbitrary imperative UDFs and formally establishes necessary conditions for reordering fixed-signature operators.The problem is addressed at the level of data-flow programs rather than fully specified relational algebra.
  • Contributions: Static code analysis derives the knowledge needed to reorder UDFs without requiring their complete algebraic semantics.The analysis operates over imperative UDF implementations to estimate properties needed by the optimizer.
  • Evaluation: The Stratosphere implementation reproduces most relational optimizer reorderings, including join and selection reordering and some aggregation push-down.The study evaluates the implemented concepts in Stratosphere.
  • Evaluation: The system also finds optimal plans for non-relational tasks without being told operator semantics in advance, and the approach applies to parallel data-flow systems using imperative UDFs.This extends optimization beyond relational data flows and beyond Stratosphere itself.

2. BACKGROUND: STRATOSPHERE

Stratosphere expresses programs as data-parallel DAGs whose operators combine second-order PACTs with first-order UDFs. Its programming model includes record-at-a-time and key-at-a-time operators over structured records and keyed groups.

  • Stratosphere Architecture: Stratosphere separates the Nephele execution engine from the PACT compiler, which translates Java user programs into executable DAG data flows.The compiler can exploit declarative aspects of PACT programs during compilation.
  • Programming Model: A PACT program is a DAG of sources, sinks, and operators combining second-order functions with first-order UDFs.Second-order functions partition inputs into groups and apply UDFs independently, exposing data-parallel execution opportunities.
  • Programming Model: The five implemented PACTs are Map, Reduce, Cross, Match, and CoGroup, covering unary and binary data-processing operators.Map creates one-record groups, Reduce groups records by key, and Cross, Match, and CoGroup operate on two inputs.
  • PACT Semantics: Map emits one UDF result per input record, Reduce applies a UDF to each key group, Cross applies it to every record pair, and Match retains pairs with matching keys.The formal definitions distinguish unary mapping, keyed reduction, Cartesian pairing, and keyed matching.
  • Programming Model: Map, Match, and Cross are record-at-a-time operators, whereas Reduce and CoGroup are key-at-a-time operators that process lists of records.For key-at-a-time operators, records sharing a key form a key group.

3. A REORDERING EXAMPLE

The example shows how partial knowledge of UDF behavior can safely identify useful reorderings without full operator semantics. Static code analysis conservatively derives this knowledge, preserving correctness while potentially missing valid data-dependent reorderings.

  • Reordering Map1 and Map2 preserves the output, and is desirable when f2 filters a significant portion of the input records.The example contrasts this safe reordering with the unsafe reordering of f1 and f3.
  • The optimizer reasons about attribute conflicts rather than whether an operator computes A + B or A · B.An operator that writes A conflicts with another that reads A, preventing a semantics-changing reorder.
  • Static code analysis estimates the properties needed for reordering by examining field accesses, conditions, and updates in imperative UDFs.The analysis identifies read-set membership from getField use in a condition and can identify field updates from setField operations.
  • The analysis is conservative: it guarantees safe reorderings but may prohibit valid reorderings that depend on unreachable execution paths.If all inputs satisfy A ≥0, f2 and f3 could be reordered, but static analysis cannot detect that data-dependent fact.

4.1 Definitions

The framework represents all accessed attributes in a global record and characterizes operators through read and write sets. Reordering relies on the readonly conflict and key group preservation conditions.

  • The global record uniquely names every base and intermediate attribute, while a redirection map links dataset field indices to those names.This prevents reordered operators from interpreting static field indices as the wrong attributes.
  • An operator’s write set contains attributes it may create or change, whereas its read set contains attributes that may influence its output.Read-set membership is defined by changes in output cardinality or output values when one input attribute changes.
  • The readonly conflict condition requires that neither operator reads or writes attributes changed by the other, and that their write sets do not overlap.This condition captures noninterference between the operators’ attribute dependencies.
  • Key group preservation requires a UDF to emit one record per input or consistently emit or filter records within relevant key groups.The condition is additionally needed for reordering key-at-a-time operators.

4.2 Reordering MapReduce Programs

The paper proves sufficient conditions for reordering Map and Reduce operators by combining readonly conflict with key-group constraints where aggregation changes group structure.

  • Two Map operators can be reordered if their first-order functions satisfy the ROC condition.The proof establishes equivalent outputs by showing that the operators preserve the relevant attribute values and execution paths.
  • ROC alone cannot guarantee Map–Reduce reordering because reordered plans may create different Reduce key-group cardinalities and therefore different outputs.The additional KGP condition addresses this problem by requiring whole key groups to be preserved or filtered.
  • A Map and Reduce operator can be reordered if ROC holds for both UDFs and KGP holds for the Map UDF with the Reduce key.KGP preserves the key-group cardinalities needed for the Reduce UDF to process corresponding groups equivalently.
  • For two Reduce operators, reordering requires ROC plus KGP for both UDF-key pairs.The proof follows the same correspondence argument used for the Map–Reduce case.

4.3 Reordering Binary Second-Order Functions

The section derives reordering conditions for binary PACT operators by conceptually reducing Match and CoGroup to Map, Cartesian-product, or Reduce forms. These transformations expose conditions based on operator read/write sets, grouping preservation, lineage, and key structure.

  • Common operator transformations: Binary PACT operators can be transformed into Map operators over Cartesian products, reducing their reordering analysis to a common case.Cross operators already apply a UDF over every pair, while Match operators incorporate their implicit equi-join into a transformed UDF; these are conceptual, non-intrusive transformations.
  • Map and Cartesian-product reordering: A Map can cross a Cartesian product when its read and write sets do not intersect the other input’s attributes.The condition is (Rf ∪Wf) ∩S = ∅, with the symmetric condition applying when pushing the Map in the opposite direction.
  • Match reordering: Two Match operators can be reordered when their transformed UDFs satisfy ROC and neither operator reads or writes attributes from the other operator’s unmatched input.The formal conditions also add each Match key set to its transformed UDF’s read set.
  • Reduce reordering: Reduce reordering requires preserving groups, typically by including the Cartesian-product input in the Reduce key and ensuring the Reduce UDF does not use those attributes beyond grouping.Under these restrictions, Reduce can be pushed through a Cartesian product; special Match cases can add attributes to the Reduce key when foreign-key structure preserves group identity.
  • CoGroup reordering: CoGroup is transformed into Reduce over a lineage-tagged union, enabling reordering when UDF behavior can be restricted to the relevant input.A lineage attribute distinguishes records from the two inputs, and a Map can be pushed below the union when it uses only one input’s attributes.

4.4 Possible Optimizations

The derived conditions support relational-style join and selection reorderings and limited aggregation push-down, while excluding rewritings that require semantic information such as associative side effects.

  • Supported optimizations: The conditions support the full set of join and selection reorderings considered by relational database optimizers, plus invariant grouping as a basic aggregation push-down.More advanced group-by transformations remain limited for arbitrary UDFs because they require knowledge of the aggregating function’s nature.
  • Limitations: The optimizer cannot establish reorderings that require semantic information, including associative side effects, and static analysis imposes additional restrictions.For example, it cannot reorder two Map functions that add a constant to the same field.

5. DISCOVERING PROPERTIES VIA CODE ANALYSIS

The code-analysis method estimates the properties needed for safe reordering from Java UDF implementations. It uses conservative control-flow and data-flow analysis to over-approximate reads, writes, records, and output behavior.

  • Analysis targets: The analysis estimates global records, read sets, write sets, and output cardinalities needed by the reordering proofs.Read and write sets are derived from UDF code, while emit cardinalities can be estimated by traversing the control-flow graph.
  • Analysis framework: Static code analysis operates on Java bytecode using control-flow graphs and use-definition and definition-use chains.The assumed UDF representation is typed three-address code with record, variable, branching, arithmetic, and function-call operations.
  • Read-set estimation: Read sets are estimated by locating statically computable getField accesses and mapping field positions to attributes of the global record.The method assumes getField is the record API’s access mechanism for individual input fields.
  • Write-set estimation: Write-set estimation tracks emitted output records and their constructors to account for explicit and implicit copying or projection.When different constructors occur on different paths, implicit projection is chosen conservatively.
  • Safety: Safety follows from conservative supersets of the true properties across all execution paths, so every generated reordered plan remains equivalent for every input.The analysis adds attributes when uncertain, trading precision for a safety guarantee.

6. PLAN ENUMERATION

The plan-enumeration algorithm recursively generates all data flows obtainable through valid pairwise reorderings, using memoization to avoid redundant work. Its search space is constrained by the initial flow’s structure.

  • Enumeration strategy: The algorithm enumerates valid alternatives through recursive subflow enumeration and exchanges of reorderable neighboring operators.It differs from traditional relational enumeration because its input is an existing data flow rather than an algebraic query expression.
  • Recursive enumeration: The algorithm appends the original root to recursively generated alternatives, then tests candidate roots for reorderability and recurses after replacing the root.In the example, Map3 can exchange with Map1 but not Map2, producing only reorderings allowed by the pairwise conditions.
  • Efficiency: Memoization and restricting recursion to distinct root candidates reduce duplicate enumeration and recursive runtime.The memo table stores alternatives for previously processed data flows.
  • Physical optimization: The enumeration can integrate with a Volcano-style physical optimizer by retaining the least expensive plan for each interesting property and root candidate.The adapted procedure must return at least one plan for every possible root of a subflow.
  • Limitation: The approach cannot change plan decisions already implied by the initial data flow, including some circular join graphs.This dependence on the initial flow distinguishes the method from relational query optimization.

7. EVALUATION

The evaluation tests the optimizer on relational OLAP, biomedical text mining, and clickstream data flows, finding broad plan-reordering coverage and substantial performance differences between plans.

  • Experiments: The least-cost TPC-H Query 7 plan runs in roughly 6 minutes, while the last-ranked plan takes about 45 minutes, or 7× longer.The enumeration explored 2518 alternative plans, and the least estimated-cost plan also had the shortest execution time.
  • Experiments: The best text-mining plan outperforms the worst selected plan by almost an order of magnitude according to estimated cost.The task uses a Map pipeline with selective, compute-intensive NLP components whose dependencies constrain valid reorderings.
  • Experiments: The best clickstream plan beats the implemented data flow by a factor of 1.4, demonstrating optimization of non-relational Reduce operators.The task processes 430 GB of click data, 13.8 GB of login data, and 9.2 GB of user-information data.
  • Optimization potential: The optimizer explores large fractions of conventional relational search spaces while also enabling optimizations unavailable in current data analysis systems.Evaluated rewritings include bushy join orders, pushed aggregations, and reasoning about interesting properties; the clickstream task demonstrates non-relational reordering.
  • Enumeration time: The prototype’s enumeration faces exponential search-space growth and does not yet use cost-based pruning or efficient plan-enumeration techniques.The paper identifies search-space pruning and enumeration overhead as future work.
  • Feasibility of static code analysis: Static code analysis enables enumeration of almost all valid plans for the four evaluation data flows using automatically derived read and write sets.The prototype obtains UDF information through annotations or static code analysis before enumerating alternatives and applying physical cost optimization.

8. RELATED WORK

Prior work optimizes restricted UDFs, higher-level algebraic specifications, or runtime configurations, whereas this work directly reorders data flows without assuming operator algebraic properties.

  • Static analysis: Manimal’s static analysis and index or compression optimizations are complementary to this paper’s operator-reordering approach.The paper identifies Manimal as the most relevant related work but distinguishes its optimization target.
  • UDF optimization: Existing extensible-RDBMS work considers UDFs with relational-selection semantics, leaving the challenge of identifying reorderability largely outside its scope.Those systems focus on when reordering is beneficial rather than whether it is possible.
  • Algebraic approaches: Higher-level systems translate algebraic specifications into data flows, while this work optimizes data flows directly without operator algebraic knowledge.The contrast includes systems such as AQL, Pig, Jaql, Hive, Tenzing, DryadLINQ, and SCOPE targeting parallel platforms.
  • Runtime optimization: Starfish optimizes Hadoop job configurations using runtime profiling rather than inspecting or reordering the program itself.Its approach is therefore distinct from source-level data-flow optimization.
  • Code translation: Ferry translates general-purpose application code into SQL to push processing instructions into a DBMS, following an algebraic approach.This differs from optimizing arbitrary data flows without requiring algebraic operator properties.

9. CONCLUSIONS AND FUTURE WORK

The paper shows that static analysis of black-box imperative UDFs can support substantial data-flow optimization without known algebraic semantics. Its Stratosphere prototype reorders relational and non-relational flows, while beneficial ordering and broader optimizations remain future work.

  • Static analysis of a handful of properties enables filter and join reordering and some aggregation push-down for black-box imperative UDFs.The approach avoids requiring a full algebraic specification of operators.
  • The optimizer formally establishes reordering conditions, estimates required properties from UDF code, and enumerates plans without using algebraic properties.
  • Stratosphere experiments reorder both relational and non-relational data flows, improve runtime by up to an order of magnitude, and perform optimizations unavailable to algebraic optimizers.The experiments also find that static analysis extracts the properties required for reordering.
  • Future work will determine which reorderings are beneficial by estimating black-box operator selectivity and execution cost.Planned extensions include broader plan-level, semantic, and code-transforming optimizations.
Loading 1208.0087v1…