[Proposal] Allow Spark to act as the execution engine for druid queries

[jisookim0513]

Currently, Druid has a rough time when there is a long query that prevents from other shorter queries to be processed. In order to mitigate the effect of long queries, I'd like to enable Spark to serve as the execution engine for druid queries. I do not have a concrete (or set) design of what I would try to implement, but here are the two approaches I am considering:

1. Make Spark query historical nodes and merge the results like a broker does.
2. Make Spark download segments from S3 and do all the computations by itself.

Links related to this issue/proposal:

https://groups.google.com/forum/#!searchin/druid-development/spark/druid-development/ULdKYZeven4/NIM5ySTbAQAJ

https://groups.google.com/forum/#!msg/druid-development/Jp9Lv0mGFlg/f9m1U9MBBAAJ

Related PRs:

• refactor ClientQuerySegmentWalker #2837 refactoring of client segment walker

[jisookim0513]

Quick updates on the progress:

I now have a working prototype. There is a service that maintains a server view and returns a snapshot for the given query, which is similar to what #2282 does. And using the server view snapshot, Spark driver submits a Spark application that does something similar to what a Druid Broker does. If #2282 gets merged, I plan to eliminate the server view on my side and query Broker for the snapshot instead.

What the Spark driver and application do is to chunk the query into multiple parts (segments) and spawn multiple intermediate-level brokers that process a specific number of segments. And then the intermediate results get merged by the final stage broker. There is only one level of intermediate merging for now, but I might make the number of intermediate levels configurable. The number of segments that are queried simultaneously is configurable so I can control how many segments are queried at once to the historical nodes, instead of querying everything at once. It takes more time to get the results back from the spark-druid project due to the initialization time of Spark application and each Druid broker. However, query failure rate is very low compared to querying Druid broker (no OOMEs and timeouts).

Currently, Spark drivers are run locally. Since Spark requires (or highly recommends) each SparkContext to be on separate JVM, my service spawns a new Java process for Spark Driver each time it receives a new query. This is fine for testing purpose, but not too good for production. My next step is to integrate it with Mesos so Spark driver for Druid queries can be run on Mesos and I wouldn't have to worry about managing resources.

[gianm]

@jisookim0513 it sounds then like you went with approach (1)?

[xvrl]

@gianm indeed, we started with the first approach since most of that work is a pre-requisite to get the second one working. So far that seems to solve our use-case, so unless we have issues with it, we probably won't go down the second option.

[himanshug]

I wonder if we could define a new type of node, say "tiered-broker" which splits input query (by query interval) and forward the splitted query to regular druid broker, then merges (probably simply concatenates) results from downstream brokers and streams the whole resultset to user.
That would be same approach minus spark and mesos.

[drcrallen]

@himanshug such a thing would also be handy for large number of peons.

[gianm]

@jisookim0513 thanks for the update, this sounds awesome! Is there ability for a user to skip the final merging and just get an rdd that one could do any further work on?

@himanshug I think we could do tiered-brokers but that's separate. IMO getting the data into Spark is nice in and of itself, since it means a user can do further processing there in a distributed way, instead of streaming results out directly to/through a single machine.

[jisookim0513]

@gianm sure, I will have to refactor the code but you can get an RDD consisting of query results which would allow further processing such as join (although the result might not be accurate in some cases).

[gianm]

@jisookim0513 sounds cool, what do you mean by "might not be accurate in some cases" though? Is that just because you might be using approximate stuff like topNs, or is there something else going on?

[jisookim0513]

@gianm yeah, topN queries might not have accurate results.

[gianm]

@jisookim0513 Okay that makes sense thanks. do you have public code for this anywhere?

[jisookim0513]

@gianm not yet :)

[navis]

I've made hive storage handler via approch (1). Because current hive does not provide opportunity to change DAG in planning stage(which is possible in Spark), I've used select query with filters and projections. The bottom line is that it was too slow. If hive takes 100msec for processing n rows, historical node can make them in less than 50msec. But json serde + network takes 100~300msec which ruins overall performance. it's sub second for single query but retrieving lot's of rows via hundreds of queries(1G tez task throws OOM for 3000+ rows) was not the right way to be done. I have some ideas to improve this but it's low priority job and seemed not be done in a month or two.

ps. But still my colleagues like it because simple select SQL is not ran on MR or TEZ, which means the result is returned in sub-second, of course. Some uses it loading druid datasource into hive table by using CTAS or insert query.

[gianm]

@navis that sounds great, is that public? Have you looked into making groupBy queries (when appropriate) in addition to select? That should really take advantage of Druid's abilities better, and limit serde/network traffic.

btw, what were the ideas you had to improve the performance of "select"? I'm curious because IMO improving "select" and "groupBy" are both pretty valuable, they are used a lot by users but not very optimized compared to "timeseries" and "topN". I did some work on groupBy in #2998 but have not looked at select at all yet.