Cognifest NYC 2017: Yuri Bogomolov, Distributed Time Series Analysis: Before and after Spark SQL
wow [Music] hello uh my name is city and today i'll talk about justification series analysis my talk is very technical this is why if you have any questions please feel free to ask them i have a time bubble okay i could disclaim disclaimer from our compliance teams his top contains no investment device nine approvals okay so what is a time series time series is a series of data points sorted by time some examples of time series data you might think of are stock market prices temperatures oil prices and the list goes on and on time series data can be used to predict or forecast the future values based on surpassed observations for example we may be able to continue tomorrow's quality prices based on what has already happened we may be able to use our intuition on multiple data sets and join them together for example we may believe that temperature in the area plays an important role in the way how horn is priced and if we join these two time series together we can build a better forecast let's recall what is left join that's right it's a joint function set is based on exact matches if we don't have an exact timestamp match in the second time series our result table will have null values on the right side temporal join is based on a matching criteria over time let's look at the look backward join first we are trying to find a data point with exactly the same timestamp but if it is no data then we look for the most a data point and this is the value we use and it's very similar for look forward join let's look at one quick example so we want to join two times three data sets and the result table will contain the same timestamps as our left so it's kind of similar to left line for 8 a.m we have an exact match of time in both data sets so this is the data that we use for 10 am we don't have an exact match inside data set so we fall back into two dollars and for 12 years again we don't have inside much so we look for the most decent role which is three dollars and in titus we have thousands of market data sets and a lot of data sources with unaligned timestamps so we need fast and scalable distributed technology first we started investigating existing solutions and it turned out that the existing packages either can't handle huge data sets or they don't support the joints that we need so we decided to build a new time series library on top of apache spark and we achieved that by building a time series aware of data structure that extends rtd and optimizes data locality what are the main differences between our time series lcd and regular opportunity from spot we associate time range on each partition and also we track constitutional time ranges for all populations and finally we preserve temporal order every operation will output a source of time series again let's take a look at one example with molecular spawn obstacles you don't don't have control over positioning so the data can be prostitution in any way but with time series rct we assign time ranges to each partition and this structure helps us to implement some operations very efficiently a very common function for time series datasets is group we may want to group those with the same timestamp or we may want to group those within the same time each one let's look at the first example how would you bring those with exactly system timestamps using rctt or later fame api you would probably use group y and when you use group by actual exposure use the hash partitioner on the circuit and it shuffles the data around and then you want your output to be exhausted by time so you need another shuffle right to solve itself uh this is not efficient let's see what we can do with time series rct representation well we know that it is already sorted by time so we can just use extreme algorithms to group those together inside each partition in fact this iteration is local because all rows with the same timestamp will belong to one partition so we need zero shuffle and we need constant additional memory we benchmarked this functionality and as we expected our approach worked much better let's get back to capital join to be more specific it is defined by two parameters direction move backwards or look forward and window we talked about direction previously but what about window decides how much to backward or look forward so let's say we have a smaller window and then we would not find and watch for our time stamp from select but if we increase the window size then you will find this divine personality at the same time you don't want your window to be too weak because the data that you join should develop so this is what you need to define a temporal joint direction and window let's look at one example so we want to join these two time series data sets how can we do that just on one machine watch single port implementation you are probably familiar with the sliding quinton ltds or the merged source alphas and we can implement something similar so we can iterate through both data sets with two pointers remove the left pointer every time and remove slide pointer when there are more decent those available on tonight i'm sorry i need to do a quick sound check uh we lost the sound on one of the mic so one sec i'll just need to readjust very quickly sorry about that there's a lot of interference here so therefore yeah so you went from rdd to data frame so that was substantial you know performance achievement and then you went i'm serious why didn't you go to what time seriously uh my second part it's the second part of the process absolutely what version of spark is this we did this benchmark last year so i believe it was 1 26. yeah like my second part performance optimizations come on this is a mama all right all right uh we're good to go good okay so we started here yeah so we can just iterate with these two datasets with pointers and uh joins us again so we have a single core implementation but what about the distributed pressure let me advantage is that we assign a time range to each partition and actually once we have time ranges we can build a dependency because if we have time ranges and we have window in this case it is one hour we can not but we know exactly what dependencies what partitions need for data and once we have dependency graph we can use exactly the same streaming algae from the previous example but without it on every position and this is what we get so in this case we have two partitions so we managed to parallelize this work between two machines in this case and this joint is not locked so it still needs to shuffle some data but because we have the dependency graph we can optimize data dependencies and again our benchmark for join shows that its implementation is quite efficient but as we just discussed five minutes ago we saw this significant improvement of data names and our first time series of implementation was base and rtd 100 percent but in the future you wanted to take advantage of this data frame optimizations to be specific data things have two main benefits they have custom memory management aka project content and optimized execution plans okay contract catalyst um the top chart shows uh memory usage when caching two representation of the same data and you can see that the data set the presentation is roughly five times smaller and since bottom chart shows run time of a simple aggregation so our goal was to take advantage for this improvement why not just use data names yeah will you uh just have this point like why can't we use a build something like time series date let's compare both interfaces the rtg interface actually if you want to build your own rpg all you need to do is implement this compute functions it takes a partition and it's an iterator so in this case you have full control over the dependencies positions and you can do whatever you want in conscious data frame takes two parameters simple context and logical plan and logical plan is kind of a it's very similar to a past simple statement it doesn't provide you full control to manage dependencies but our life that we claimed is based on managing data dependencies so our goal was to get the flexibility of rtd interface but at the same time take advantage of data training um let me get right to the solution and then explain how it works our new time series rtc implementation is based on two objects we have a regular database and we have a partition in for us that contains partition metadata that we need uh how to implement some simple operations let's say how to delete columns well you can just use uh zia job function to delete columns from the underlying data thing and this iteration doesn't affect data partitioning so you can simply reduce the partition info object this was simple but what about more complicated applications like join now we have two data things but we can do the efficient temporal joint only with obstacles because again we need access to politicians and dependencies so first we need to convert data frames to our rct simulation then we run the join and then we need to convert our rtd back to the data frame okay this seems to be working oh do you have any questions uh this seems to be what you find but this is very bad design and you need to convert because rdp structure is absolutely different than data frame structure in fact that's over here that's column area i mean those conversions will take enormous time yeah but and this is really i will need to get your mic is very extensive from database to rtt because this data sets have a very different internal representation and what if i tell you that we achieved almost free conversion from one presentation so yes you're right that we have a lot of conversions going on here and actually because conversion is so expensive it may be even more expensive than the operation itself this is why our main goal was to optimize conversions and what is the cheapest possible conversion an old content right so we decided to use internal data frame rows inside our affinity and this way we can go back and forth between these two structures with zero conversion uh and this gives us you know very cheap conversions but also it gives us some positive benefits as you know travel objects have plenty big memory overhead every object has a 8 byte header and one notorious example is the fact that a string abc takes about 50 bytes of memory things are getting a little bit better in java 8 it's used memory overhead by office tuning by 8 bytes i believe and channel 9 promised even better results but the overheads are still significant and if you have a genetic code then every element of your own is a separate java object so all these overheads add up but data frames under the hood say uses unsafe router presentation so they have one one white array and they pack all these values into one byte array and it has multiple benefits first of all now you can avoid married overheads also it is real it is very helpful for garbage collection because instead of five projects you need to track just one object and there are many other applications like uh calculating a hash code making a copy of a row comparing to those all these alterations can be done very efficiently with unsafe flow representation so we started using internal catalyst unsafe roads inside our flint library but if you look at a catalyst now with once we have our structure how can we create a data thing there is no public entry points that take internal pros so we build a new entry point that goes directly to the logical plan stage so we create our own logical plan that iterates through our rct with internal dose then we expect catalyst to apply all systemizations and finally when you have a physical plan before internal rows are converted to external pros we need to get it back so because we want to work with efficient unsafe laws so we get our representation directly from the physical plant i benchmarked two real datasets once it contains only numerical data and once it has mixed numerical and streaming data and you can see that just migrating from genetic close to oxygen always gives us a lot but also when you have a data plan and when you cache a data thing rows are divided into batches and every budge is converted into a columnar representation so with this uh acknowledge we can get all these uh data frame advantages for free because under the hood we have a data frame and you can see that conversion from rtt offensive rows to a corner with this simulation gives us another significant improvement and finally data frames actually by default when you cache a data plane they will try to compress the latent and if you have just numerical data you know it's not really you can't compress it a lot but if you have many short streams like in this case the compression is also quite helpful so we have made a lot of progress with memory footprints now our time series rdd has the same memory footprint as a data frame and in case of any improvements uh just yesterday i was talking with wes and he mentioned that spark 2.3 is actually going to use arrow under the hood for the column of the presentation and we will get all these improvements for free because you just have a database but what about catalyst optimizations uh well if you look at this picture uh all the applications on the electricity level are a black box for catalyst it cannot optimize that this is why the rest of my talk will be focused on the new feature of spark 2.2 called spark session extensions the new api allows you to provide your custom logic and extend different phases of catalyst pipeline they had a limited version of this functionality before but now it was experimental and they had a limited support for it for example you can inject a possible password and extend sql credits or you can create your own optimizer rule that takes advantage of some data properties since your data sets have let me give you an example uh we have very simple code of this data pane read source by id then we select two columns and we'll expose it again i have a question to the audience how many times spark will shuffle and source this data any ideas the shuffles [Music] so if you have a large data set that will involve both the sword in the shell okay uh but i don't want to realize that i don't call dot count or something oh that's true if you don't have it you don't do a collection it doesn't sort it doesn't sort until you yeah yeah okay so if i do sort a name dot com it's like how many times spot will shuffle the data and solve shuffle will be inside the sword okay so let's look at catalyst plants that we get here first catalyst will build a fast logical plan and you can see that it this much duplicates our code we have a sub node same project now and set another social notice then this plan will be optimized by catalyst and here we can see the optimization that is called column projection uh so if you use only a fraction of columns it makes sense to get into other columns as soon as possible right so this is what smart builder will do here and finally when this logical plan is converted into physical plane spark will add an extra node to shuffle oscillator because in spark sword is represented by keynotes one node for the shuffle step and one node for sorting data locally from plasticians so we can so the right path was one shuffle two swords we can see that catalyst was smart enough to identify output partitioner of this node and we have just one cycle because catalyst knew that he doesn't need another shuffle that consists so i of serve it just once i mean it will not sort of twice already so stuff uh it's a physical plant so it will run it will sort it locally on politicians it will not shuffle the data twice but it will run it's a local sort two times but but what is the point to a second time sort of already sort of data it will not do anything right but you don't always have full control over your database you might get a data frame from somewhere right and it might be sausage or maybe not you don't know yeah but but in this case it's already sorted we see that it will be sorted first time why would it be sorted second time on something or that already sort of does yeah this is a good question for people's developments why they don't have to maybe there are two separate uh partitions maybe the depiction two separate start and then you highlight yourself yeah but like i i benchmarked it like two days ago and yeah when you have these two soft notes like this hot now because it's local it should stay locally on every not very slowly shuffle data shuffle takes uh significantly low yeah but still it takes some time okay and now let's make one tiny change to our code i highlight this right here so we do exactly the same but now we also rename a column let me ask you the same question again how many times spark will shuffle and soft today two and two of two and two yep uh so we can see supports logical plan on the left it looks pretty much the same the optimized logical plan also looks very similar but again we can see uh we can see cold projection optimization in place so our project node was split into two we see column projection here and this one will just remain our id column into ic2 and then when this logical plan is converted into a physical plant we see two shuffle steps here and this happens because spark treats id and id2 as two separate expressions id2 is an ls called id but they have different expression ids so spark doesn't know how to delay sand with each other okay so let's say we have this problem how can we solve it uh one idea that we had before was to avoid thoughts in the same data set twice and yeah it is a fair idea right but in some cases we don't have control over all code bases we may get a data frame from somewhere and it may be softer but maybe not what can we do with it and one possible solution we can achieve with the new api so we can create a new optimized group so let's see i here i have implemented a simple optimizer tool just for this problem so the most important part here is function applied it takes a logical plan and it returns an optimized possible way and every rule that is a logical plan g so it goes from that from the top population down to leaves and if there is a nodes that satisfy our platform function and presentation this will replace the additional node with our geological plane so this is the part where we are looking for these kind of structures where we have sourced the protect and another sort and in practice it will look like this so this tool will try to find the following pattern it will find a source node that contains a project nodes it contains another source node and we know that if our middle node remains a column we don't actually need to source this status again so our optimized plan will look like this and once we have this rule it's very simple to inject it into catalyst so when you create a spark session you can just use squeeze extensions to add your own optimizing tool and this is updated picture so our optimized logical plan looks much simpler and when it is converted into a physical plan we get only one shuffle and one soft so this is what we wanted all right i try to find examples that solve more complicated problems but unfortunately this new api doesn't have a good documentation and i i don't see like wide adoption of it so i i couldn't find any real examples that take advantage of this api but i want to show you that it's actually very flexible and it gives you a lot of uh different ideas it enables a lot of different ideas that you can try um first let me ask you what is missing in spark equal how much time you have like statistics okay this is one thing what else what common sql functionality is not supported in spark sql is that going oh what what's that an anti-join well yeah this is this can be implemented with extensions yeah um transactions yeah okay i personally miss indices if you worked with big tables sql entrance then you you probably know that indices defines the performance of your grades right and spot doesn't happen so technically what i said that's missing statistics it's even worse right because not only do not have the independence you have the not have the sorted column projection but you don't have the distribution statistics for efficient joints which is the real counter yeah regarding index between distributed data and columnar data you really need index uh and in memory good question yeah actually if you look at the spot documentation they say we don't need interest so much because you can cache a data set but it means that if you have a data set with one billion rows and you have a simple activation on top of this data set you will need to iterate through all rows again and again for every operation so when you don't have indices all your activations are in the best case say take lineage time so there's no sublinear computations without indices so i even go one step further right i mean the whole catalyst is optimized but you can't actually do chord optimization in the absence of distribution statistics yes so at the moment it's still kind of small so i actually have a pet project that i'm working on in my spare time and i just want to show you that the extension guide is quite powerful that you can even intellect use and due functionality to spark sql so how can we implement uh indices using systems so first we can play in your logical plan class and you could you actually you could uh do the same before you just subclass a logical plan but now you can also implement the solution and optimizing tools so whenever you can take advantage of an index you can have an optimized role for that that you know is looking for your new logical plan and then optimizes the execution plan and finally you need to contact a strategy that converts the neological plan into a physical type uh yeah it's actually a little bit more complicated than that so the prototypes that i mentioned about uh it's it's quite flawsome so uh i don't have any problems with missing energy points it's just you know it takes a lot of time and code to to have a good version of it but i'm actually i'm going to publish it like probably in 2-3 weeks once once i have a good question some links if you work with time series data on scale please check out flint and if you are interested in sparks extension samples or the prototypes that i mentioned i have a personal depository for that and please feel free to email me with any questions uh any questions um so the data is applied yeah microphone sorry raise your hand while i has a question great talk thanks um so the data is partitioned by time in the time series rdp can you give us any insight into how the underlying data is stored do you is it is it better for the underlying database to be partitioned as well or do you just load it in from something like part k and then read yeah actually works perfectly with sort of data so uh if you save a data thing into a 4k file and then load it back you you get the same partition so it's fine my second question was i think in some of your examples you showed data that was time series data is down to the resolution of a minute did you ever look at data sets for resolutions down to like milliseconds or even smaller oh actually flint by default uses nanoseconds other questions [Music] i was just wondering if you uh were able to explore using the data set api uh as a sort of hybrid between the rdd flexibility uh dataframe is essentially a data set of rows right uh we don't have fixed schema time series quality might have any number of codes this is why we actually want data frames and data flames they have it in terms of performance they have the same you know it is the same codes that last both datasets the difference is in one case you have fixed schema like if you have an object so you can use a data set for this object but in our case we just have 10 2100 phones and you may add a new column to your data plane and this was the use case that we wanted to help anyone else okay i'm sorry hand up still so you mentioned about indexes before so did you create a jar for a new feature for indexes really ah that's coming not really i actually created a twitter for uh you know this weird inheritance that i i showed here and i didn't get that much replies you know um and if you if you come up with a like bigger project like indices you know like database say they even mentioned this in the documentation that uh they don't see indices too important at this moment so um any other questions yes back to the dock um they just wanted to know so we have a legacy system that uses part 4.6 so we can use the the library are they on spot on our smartphone.6 and then we use the data frame uh api as well or just use the rdds uh unfortunately it's the latest blind version works only with spark 2.0 and or higher but i believe that if we check out one of previous videos then you can get a working flint annotation spot 1.6 because this is what we had internally uh some time ago uh and it's the second so can you use the data api with powerpoint sets or yes you can because data frames they were introduced in part 1.6 but first they were a part of the experimental api and since spot 2.0 data frames any other questions great thank you very much uh [Music] you