Recently our big data team was approached by one of our clients – a company wanting to minimize costs of analytics of its sales data. The client had an analytic team working with the Microstrategy BI tool . The tool fetches data by running SQL queries on an underlying Teradata DB . As Microstrategy can work with various data sources, the question was whether it’s possible to replace the costly Teradata DB with an open-source solution that would have similar performance.
They isolated a sample SQL query generated by Microstrategy to fetch data that took Teradata DB (Enterprise cluster) 5 seconds to compute. This time became our reference point and our job was to make it faster with open-source (even though we found out later that the actual time for Teradata was around 10 seconds).
The query operated on three fact tables, daily_stock, daily_waste, and daily_sales. Each of these referenced three dimension tables, calendar, product, and store. The query fetched information about each product in the specific product group for a specific store. It joined the dimension tables with the fact tables, aggregated facts using various filters, then joined the aggregated facts together. Each result row consisted of information about product and store, together with a count of sold and wasted items, followed by the number of products in each store’s stock for each date. This was not the optimal query, but we took it as if it was because it was generated automatically and the underlying engine was responsible for optimizing it. The structure of the data is presented below.
Assuming that data is partitioned by day, the query operates only on a selected time range, so we may ignore historical values from the fact tables.
All in all, around 700M rows are queried (first filtered, then joined together and aggregated).
We prepared a list of possible solutions that we wanted to test locally. We wanted to get familiar with the technologies to check how performant they are and how they compare against each other.
We limited fact tables to only 2 of the 5,000 stores in order to make data ingestion as smooth as possible.
The benchmarked query was kept as it is, but sometimes the syntax had to be adjusted for a specific solution. The obtained times are the best values observed out of a few tries. The presented values themselves are not important: what counts more are the differences between them.
MySQL good no 0.6s Very fast but could not handle this amount of data.
Spark good yes 7s Well known and available. Promising, but needs further testing on a multi-node cluster and more data.
Hive good yes 20s Not fast enough.
Presto good yes 5s Widely used, can query data from multiple sources with reasonable response time. Promising but needs further testing on a multi-node cluster.
Hive + Druid weak (new) yes 2.5s Very fast and scalable, but not widely used yet. Hive and Druid integration is still under development; we may reevaluate it with a new version of Hive.
The best performance of a scalable solution was obtained from the “Hive + Druid” setup; however, we encountered some known bugs and the solution doesn’t seem to be ready for deployment in a big organization yet. Standalone Hive turned out to be much slower than the rest, so we decided to select Spark SQL and Presto for further testing.
We decided to use Google’s Cloud Dataproc  as our testing platform; it allowed us to cheaply store data and create Hadoop clusters of any size on demand, with preconfigured Spark, Presto, and Hive.
In total ~700M rows, the same as for the original query. The tables in sum have ~250 columns. Tests were done on all data, using full schemas with all columns; here, for simplicity, we present only schemas which contain columns that were used by the query.
The amount of data is roughly the same as that used by Teradata to compute the test query. Assuming that data is partitioned by date, adding more historical sales/waste/stock data should not affect the performance of the query as the additional data won’t be accessed. Therefore, the presented results may be compared directly with Teradata.
Upload csv data of format csv/table_name/*.csv (there may be multiple csv files in the directory; in our case these are product, calendar, store, daily_sales, daily_waste, daily_stock). We cannot include our test data here, but you may use any type of data as long as you define a proper schema.hql and query. Blog post http://tech.marksblogg.com/billion-nyc-taxi-rides-presto-dataproc.html mentions taxi ride data which can be used instead.
Upload both files to google storage.
When we add gs://5spoc-benchmark-data/schema.sh as an init action to the cluster start command, the schema will be loaded automatically during cluster initialization, using the script that we prepared.
Create a small computing cluster in a default project (with the prepared init action) in order to convert the data.
Now log out and kill the current cluster (if you want bigger one); all data is stored on our google storage and we may create clusters of various sizes to test our queries.
Spark is configured by default on each Dataproc cluster, so we may use the same command as before.
We can create a cluster and directly query tables which we previously stored on Google storage. However, in order to achieve slightly better performance we will use the tables we copied onto the cluster’s hdfs storage during the initialization of the cluster. These are already defined in our schema so we may reuse the schema.sh init action.
copy_data_to_local_hdfs.sh init action will copy data from google storage to the cluster’s hdfs during cluster init.
Start a cluster of the desired size, select schema.sh and copy_data_to_local_hdfs.sh as init actions.
You may fire sql queries using tables that you’ve created in schema.sh. Tables without ‘_gs’ in their name have slightly better performance as they’re stored on the cluster’s dedicated hdfs.
When it’s finished we execute Presto with options that tell it to use Hive’s tables, then we may fire the sql queries just as before.
There are many more solutions that you could test; the complete list of predefined init actions for Dataproc is available here https://github.com/GoogleCloudPlatform/dataproc-initialization-actions.
We decided to include in our test the results obtained from direct Hive queries (as it’s already preconfigured), and from Spark 1.6, which you can run by downgrading Dataproc’s image version by adding --image-version '1.0' to gcloud dataproc clusters create.
The test queries were preceded by warmup queries that fetched the first 10 rows of each table (to cache the tables’ metadata). Each test query was run three times, the third time with a different product group.
The data format was chosen to best fit each particular technology (Presto performed best with ORC, the rest used Parquet). It was not optimized in any way (apart from using some custom properties of ORC, taken fromhttp://tech.marksblogg.com/billion-nyc-taxi-rides-presto-dataproc.html). No partitioning or batching was applied (partitioning by date was tested, but this had no effect on the test query as it queried data from the whole time range).
NOTE: Spark 1.6.2 does not support numeric values in group by clause, and Presto has a slightly different syntax for time range, therefore queries were adjusted to be equivalents.
NOTE: gcloud performance differs from time to time; therefore, there may be some anomalies in results but the big picture is consistent.
Time of reference query using different setups.
We opened a new account for VirtusLab company, for which we were given a $300 free trial; however, to increase the number of vcpus running at the same time we had to make a prepayment of $75, which was available to spend after the trial period expired.
All in all our tests we consumed $14.19. We ran at most 10 x 8cpu-highmem machines, but we removed clusters when nobody was using them (as they could be easily recreated with one command within minutes). The total amount was taken from our trial account.
To be able to rerun our benchmarks we kept our test data on Google storage; this cost us $1.91 per month (for 74 GB of data).
Google Dataproc platform is excellent and cheap for performing benchmarks involving various cluster sizes. If we wisely define init actions, a cluster may be ready to test within minutes with just one command, thus it is possible to conduct extensive automatic tests on multiple setups.
Open-source products can successfully compete with enterprise solutions as a backend for Business Intelligence.
Have you tried also Flink?
1. What about Hive with Tez or Spark backend?
https://github.com/…/dataproc…/tree/master/tez . The results are 70sec first query, 60sec next.
2. What about concurrency (multiple queries run at the same time), to simulate an environment with multiple agents accessing the data?
We’ve tested single query, as we took for granted horizontal scalability of those solutions (and underlying hdfs). We were aware that prod setup would require larger number of nodes. Anyway in our use case, there will be at most tens of people running queries at once, so this was not a problem for us.
3. What led you to choose Spark 2.2.0 and Presto .177 and not latest?
We’ve done those tests a few months ago and Spark 2.2.0 was the newest that time. As for the presto we’re using official google’s init action for it https://github.com/…/blob/master/presto/presto.sh… and Presto is slightly outdated in there.
4. What was the CPU and ram usage over the machines when running these tests?
Presto – based on top, as for spark, I could see presto process consuming ~400%CPU on average constantly. Memory around 5%.
@T I haven’t tried Flink. I will do it when I’m gonna have some time, it’s available as Dataproc init action.