Imagine you’re building a marketing data warehouse for your awesome Programming Deck of Cards brand. You don’t need lots of data, just some clicks and impressions metrics of your google ads, the number of mentions your brand had on Facebook posts and link all of that to your Salesforce CRM database. Kind of simple, right? Just connect to these APIs, ETL them to your database and watch them grow on your beautiful widgets.

But the reality isn’t that simple — the data is provided in inconsistent granularities, API vendors call the same entities differently, the dimensions and metrics are not exactly what your business logic implies — complexity increases quickly.

And then one day you want to add data from LinkedIn, Twitter, and your own MongoDB data, and one of your data providers notifies you of a miscalculation in the last 3 month data because of a bug.

For every change in your schema/data you enter a development cycle, which most of the times is detailed and complex, your Marketing data warehouse grows along with your development hours, and that’s never a good thing.

What we do at Datorama is take this problem and reduce it to a simple, out-of-the-box solution, so you can work on the Shakespearian version of the cards. Of course, the fact that we have thousands of customers, each with dozens of data sources, TBs of data and millions of monthly queries — make this problem so much more interesting, challenging, and fun!

At the core of Datorama’s technology stack, lies Dingo. An offspring of endless hours of coding, imperative schemas (for real!), functional creativity and obscure acronyms. Those who try to debug one of its complex queries, even armed with septuple-monitor and a slice of pizza, might find themselves staring at their screens while silently humming Jeff Dean’s numbers every programmer should know.

Dingo is a reliable beast, but, as any dog trainer will tell you, you cannot teach an old dog new tricks — at least not easily. Dingo isn’t flexible enough to accommodate many of the requested cool features, is very tricky to optimise and quite difficult to debug.

So…what can Dingo do?

In short, Dingo is a Java engine and API for running analytical queries over any source of factual data. Upon receiving a query request from a user, it starts with collecting raw data from the relevant fact tables, then does a few iterations of joining, computing, filtering and aggregating the data until the final result can be sent to the impatient user.

One of the main shortcomings of this process is that there is no separation between the plan and its execution — it all occurs at once, and as such operations such as debugging or reviewing it post-process are very cumbersome, and implementing new features or inserting optimisations become difficult.

And Pluto is born.

Pluto is designed and architected to solve all of these problems implicitly by 1) separating the plan and its execution, 2) decoupling the logical and the physical parts of the plan, all of this along with 3) improving Dingo’s distributed data processing capabilities. The first and obvious candidate to build Pluto on top of is Spark. Spark has all the important building blocks— distributed computing, clear and concise API, monitoring and instrumentation tools— plus it is JV(a)M-ish, so it can seamlessly integrate with Dingo’s existing code. In fact, Spark-SQL, which is Spark’s API for handling structured data, is even a better and more exact choice since its most basic data structure, the DataFrame, is essentially a data table — exactly what Dingo processes.

Like any big software evolution, the first step is to make sure that the direction you’re taking is not too far off — that it is possible to achieve the same functionality, with performance in the same dial code.

Calculated Measurements (the lego-kit function)

So you bought your kid (well, ok, yourself), this awesome Star Wars lego kit. All the parts are there, along with the instructions, and building it perfectly will take some time and patience. However, your manager, err, kid, wants to build an amusement park from it. So you take some pieces from it, some from older lego kits, use your imagination, cook and stir, slice and dice, and BAM!

One of the awesome features that give Datorama’s platform its unique lego-like flexibility (and customer support many headaches, since some users just can’t get enough of this power) is the ability to run a customised function using any of the fields, even if the other fields themselves are customised functions. This feature (among others), forces Dingo to run some of its calculations by itself, regardless of the data warehouse that stores it. This obviously complicates the processing, since some of our data is quite big, and cannot be computed by a single machine.

For example, let’s assume that Pete, our beloved user at sterlingcooperdraperprice.com has a Data Stream of clicks, impressions and User-Agent (i.e. Mozilla/ Chrome/ etc.), and is observing CTR (clicks/impressions) for one of his brands. Unfortunately, The Data Stream provider, “Oogle”, notified him that because of a bug in one of the browsers, (you guessed it right) Internet Exploder, each user click is counted twice.

Id  | UserAgent      | Clicks       | Impressions
1   | Chrome         | 10           | 100
2   | I.E            | 18           | 150
3   | Mozilla        | 20           | 200
4   | Chrome         | 15           | 100
5   | Chrome         | 18           | 120

Obviously, Pete would like to handle this the right way, and luckily he has Datorama in his toolbox. Pete creates a new calculated function as follows:

function "game-changer":
if UserAgent=="I.E":
    clicks = clicks/2
return clicks/impressions

Great! Pete can finally analyse his beloved CTRs, and view them on dozens of beautiful widgets!

But things aren’t that simple. Since this function must be computed in local memory, if the total size of the records exceeds it the computation will fail. So even if Pete streamed the data to one of Datorama’s heavy-duty data-warehouses, he would not be able to view all of it at once.

Lots of big data solutions’ providers encounter this problem when they wish to implement one of their secret-sauce algorithms in a Big Data architecture. However, in Datorama, the calculated function feature is one of the major differentiators we have since it essentially allows our users to write any function (besides, well, System.exit).

The fact that this feature is so central to the platform and so unique in the Big Data landscape, we decided to tackle it first.

The calculated functions feature introduces several complexities:

• A function can reference many fields for its computation
• It can reference other calculated measurements as part of the computation, meaning that the order of computation in the rows must be determined beforehand
• Some functions are very complicated

UDFs to the rescue!

The first thing that comes to mind when dealing with a user-defined-function is, well, using them. Spark SQL has a built-in mechanism for dealing with customised functions — UDFs (yes, User-Defined-Functions). However, the user in UDF references a Spark developer. As a first attempt, we’ve tried to generalise the UDF mechanism for our needs.

Although the UDF mechanism does have great flexibility when it comes to writing a specific function, its shortcomings are apparent when trying to compute a function over function, and it has a limit of 10 input variables (if you go with the docs) or more (if you do some hackery), but then you are limited to a single return type.

Solution

The solution that we found to be successful is composed of the following steps:

1. Index the fields by their order
2. Serialise the calculated functions and index them by their order, so if an expression e1 is using another expression e0, e0 will be computed before, thus will be found first on the compiled expressions index. Along with the compiled expressions, also keep their final types and names. The combined list of fields and calculated functions (their names and types) will be the schema for the final dataframe.
3. Convert the dataframe to its underlying RDD, then for each row run the serialised calculated functions in their order
4. Create a new dataframe with the fields and calculated functions’ results according to the schema

Performance and Benchmarking

We’ve tested few calculated functions that grasp the previous complexities:

1. ROW-SUM: Summation of all the metrics in a row, then each additional calculated measurement will add 1 (references many fields, references other calculated measurement)
2. GCD between two measurements (how cool is that? finding the greatest common divisor between clicks and impressions of your favourite ad?) (complex function)

Results

1. ROW-SUM: In comparison with Dingo, the Achilles heel of the Spark implementation is the row sum. Because each cell in a row is dependent on the previous cells, the computation must run sequentially (per row). Here, the optimisations of Dingo and the overhead of Spark made a significant, yet expected, difference. Dingo performed better than our Spark-based implementation quite consistently.
2. GCD: Here Spark performs better than Dingo as the number of rows increase. The difference between 500K rows and 1M rows is less than 10%, whereas Dingo performance is decreased linearly by the number of rows.

Conclusion

Well, our little POC of Pluto on Spark was pretty successful. The fact that without any optimisation, the calculated functions performed as we expected (and better than that in the case of the GCD), is both a testament to Spark’s power and Pluto’s direction.

The small amount of code that it took to implement it is another great positive point to mention, this is due to Spark’s (and Scala’s) conciseness, but that should be a different post…

Check our open positions: http://datorama.com/join-us/

Taming the Beast was originally published in Datorama Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.