sfspark.org: Akshat Aranya, Interactive Query Platform on Spark at Quantcast
Recording: sfspark.org: Akshat Aranya, Interactive Query Platform on Spark at Quantcast
need portable hi everyone thanks for coming thank you Peter so as Peter was saying I'm going to talk about what we're doing bits per can discuss one of our use cases so I'm going to talk about why we're using spark for our use case I'm going to talk about what the architecture around it and beyond that I would just like to talk about what some findings that we had with what we wanted to do so what's the problem that we're trying to solve so as peter was saying we get did a lot of data from hundreds of thousands of publishers and across hundreds of millions of websites and we have to process the data to produce roll ups for them so I don't know if you can really see but the point is that we produce data points like reach numbers page views and the demographic composition across a lot of different data points and it becomes a combinatorial explosion for which we have to do huge amounts of batch computation so as you can imagine that if you're slicing you're dicing and dicing your data in multiple ways and you're combining it suddenly you have to calculate data points like the number of data points that you have to calculate just keeps growing so this is becoming a problem in our batch computation system because we're pre computing these numbers and what's worse is that most of these data points will never even be looked at so if somebody is going to a website and looking at it with certainly not everything will be looked at in a given day it's also not very flexible because the data that you get is whatever has been pre computed which means I cannot do arbitrary filtering like finding out what is the female 18 to 25 year old audience in New York interested the people who come to my website what they're interested and maybe you want to filter further but it's not possible because the pre computed results are just whatever we think that people would want to look at but that may not be true so to solve this we came up with an architecture where we don't precompute these results instead we have semi aggregated data that we keep in no sequel stores and then we use Park to distribute queries on top of it and do in memory filtering we also realized as we were building up 40 types that not we could not our use case did not fit one particular no sequel store so some no sequel stores are good for certain things like doing range scans some of them are good at doing point queries and our use case just did not fit one of them plus most of the no sequel store generally don't even support a joint operation which is why we started looking at spark in the first place that if we could do joint operations with spark we wouldn't need that much capability from the no sequel store so like I said we join we use multiple no sequel stores and we joined it across them and on top of the nosy cool stores we have a query API which can be used by the website and it has flexible filtering capability so if a user is drilling down if they want to get hold of their female 18 to 25 audience those filters can pass down to a query a peer and we can calculate those data points so the architecture is something like this we are using edge base and aerospike for different data sets we batch load the data on a daily basis and the database data is charted 256 ways in a manner that the user query can perform parallel independent joins a user query comes in through the query API and is executed as a short SPARC program so we are programs usually run at kind of web speed so you can think something like one second to ten seconds which as I will talk about later curb made us realize some problems it's part so the user query comes in to the query API it is divided among the spark executors and each executor picks up a shard from each base and from aerospike and joins the data and we do linear scans on each base and then we do hundreds of thousands of random lookups on aerospike for each user query and finally a reduce by key operation combines the results and returns the results to the user so in this architecture we have the spark executors are long-lived so they're servicing multiple queries I mean they stay alive all the time and they're serving servicing multiple queries this is because we cannot afford to bring up executors every time a new query comes in because we're trying to really make the latency below separate from this is our cluster scheduler which is used to allocate spark executors on our cluster and that helps us in having a fault-tolerant architecture where if we lose executors the spark driver will realize that an executor is gone and then it will ask for a new one in its place so some of the things that we found out while building this was well I mean it's the sea so there's a single point of failure within spark so even though the executors our fault tolerant the driver itself is a single point of failure so if you lose your driver all the context of everything that you've accumulated in memory is also gone and the executors go away which makes Park somewhat less ideal to run long-running services it's fine for running jobs but if you're trying to build a website off of it and its long-running service that doesn't quite work and especially if you are trying to build a website of it you cannot afford to have spark executors with large amounts of in-memory state because the startup time would be too high so initially we were trying an architecture way we would just load the data off of off of the file system and just do everything all of the queries in memory and we soon realized that that was not going to work for us because that makes the worker bring up time maybe like a couple of minutes and that really cannot work for web service which is why we decided to go in the direction of using no sequel stores so that the spark workers are stateless and you can very quickly bring up new ones if you lose the driver in fact one of the architecture things that we're planning to do is have standby drivers which which will be standing by asking for executor so if we if you lose your driver son suddenly all the executors will be gone but the new driver will be quickly able to get new workers and continue from there so that kind of gives us some capability of being fault-tolerant long-running service the other thing that we found out which park is that even though it runs much faster than Hadoop there are still problems when you're trying to run really short jobs like jobs which take up a couple of seconds so as you see in this example the jobs are running let's say one to two seconds long but the initial yellow portion which shows scheduling delay is about 200 milliseconds and that becomes a problem when you're trying to do really low latency things which part we found out that most of this is happening because of Java serialization so spark has the capability of using cryo serialization but it's only it only works for serializing the actual data in the rd DS it doesn't work for serializing the tasks themselves so we have submitted some patches which allow you to use Park to use cryo serialization for task scheduling as well which helped us reduce the latency quite a bit also another thing to notice is that the scheduling time as this figure shows is proportional to the number of partitions that you have so if you think you can paralyze your work by just having more partitions that only works to a certain level because then the scheduling time itself starts becoming important and finally I have a list of other things that we learn from our spark experience so when you're trying to really speed up your spark program such that they run in a few seconds it's very important to experiment and fine-tune the number of partitions and this ties into the fact that you cannot paralyzed as much as you would like to so it's important to make the number of partitions be what you are willing to handle the amount of latency that you are willing to take so it's really a lot of experimenting with the number of partitions another thing that I observed was that the way you write your spot program really influences in influences the number of shuffle points and the number of stages that your program is going to have and you may think you understand how many stages your program has but that's not always the case so it's always good to go and look at the actual execution of the program to figure out how many stages there were how many shuffle points there were which can really help in speeding up your spark programs we also found out that using map partitions is useful in some cases so my partitions if you don't know it lets you basically it's like a big hammer kind of operation which lets you just run a certain program on each partition so you can kind of collapse the program and do everything within map partitions which in generally may not be a good idea but in certain cases it really helps in optimizing because you can avoid having needless shuffles if you can do some kind of multiple steps within map partitions another thing which was really surprising for us was in a tight loop we have our having problems with memory utilization and we realized that was just happening because of how four loops are implemented in Scala and if you're using zip with index in a for loop that's going to create a whole lot of needless objects and cause a lot of GC pauses so a simple fix for that would be using while loops instead and it sounds like a very small thing but it made a huge difference for us in our programs that concludes my talk if you have any questions I'd be happy to answer yes so this is a tool which was released by UC Berkeley it basically collects a trace of events so spark has a capability of collecting a trace of events saying that this task was started this is when it started serialization this is when it went to the worker actually started executing this is when it sends back the result the tool from UC Berkeley let takes in that input and plots it out so I found this very useful for debugging the performance yes I don't think anybody is working on that problem because nobody as far as I now really cares about scheduling delays given that TAS usually run much longer than couple of seconds but the problem is that when a job is broken up into tasks eat it basically creates a list of tasks before it starts scheduling even the first one so you can imagine that just creating that list of that and producing that list of tasks is a pretty expensive serialization work which is why cryo serialization helps over here but essentially it creates one array of task and it doesn't start scheduling them until that array is created so if it actually did create one task and schedule it it would be much faster but it doesn't do that yet no I don't think it's relate to that at all because that decision has already been made it's just just a simple way of doing it I guess yes we are hiring yes so if you are interested in finding out about this project or any other project that are happening at concast talked to me after this alright that's it thank you