Scale By The Bay 2021 : Itai Yaffe, Yakir Buskilla, Casting the spell: Druid in practice

Scale By The Bay 2021 : Itai Yaffe, Yakir Buskilla, Casting the spell: Druid in practice

Recording: Scale By The Bay 2021 : Itai Yaffe, Yakir Buskilla, Casting the spell: Druid in practice

thank you for joining us um if you ever dealt with big data you probably ask yourself how you can ingest terabytes of data and how you can make all this data available for customer facing dashboards because in today's world clients are expecting to get answers in real time and finally how you can do all of that in a cost efficient manner because organizations today deeply care about bottom line so you came to the right place because this is exactly what we're going to discuss today we've been using apache druid for the past six years now and we believe that we have some good answers to the aforementioned questions so before we start just a short disclaimer uh this presentation mostly covers advanced topics in apache druid um but we still believe that it can be useful for broader audience so no need to worry if you don't have previous experience if you are looking for an introductory session about apache druid you can check out our previous stock from big data london in 2019 and finally this presentation mostly covers our um journey with apache druid while me and i worked at nielsen so my name is jackie i'm co-founder and ceo at cocoa bi and before that i was leading the engineering for nielsen identity i mainly focus on big data and machine learning machine learning solutions and with me today itai who is currently a senior solutions architect at databricks before that he was a principal solutions architect at imply and before that he worked at nielsen as our big data group tech lead and as you can see he has been dealing with big data challenges for quite some time okay so what we are going to discuss today so first of all we're going to discuss data modeling and we're going to show you how you should model your data when you're gonna ingest it into druid then we're gonna discuss data ingestion and we're gonna show you how to ingest terabytes and petabytes of data into apache grid then we're gonna discuss retention and deletion because it is very important to know when to delete uh data from druid because you don't want to pay for data that you don't use and finally we're gonna discuss quality optimization and sharing with you some tips and tricks on how to improve the performance of your queries and of course that we're going to show you how to leverage all that knowledge in your organization okay so before we jump into the technical details uh some business context so nielsen is a data and measurement company houston is very famous for providing tv measurement tv rating uh especially in united states and essentially what nissan is doing is collecting anonymous device level and personal level data from various sources online sources offline sources panels and the idea is to create an holistic view of a consumer out of all these pieces of information that are being collected and provide insights on this consumer so this insights are being used by the different measurement and targeting products in nielsen and in nielsen we have a very extensive data infrastructure these numbers represent only a single division in nilsen but as you can see these numbers are very impressive we are running uh we are processing more than 10 billion events on a daily basis on all kafka clusters if we're talking about s3 we are storing models in 60 terabytes every day on s3 and in general the data lake consists of more than five petabytes of data um talking about spark we are running more than six 000 nodes of spark on a daily basis and finally when talking about druid we are ingesting tens of terabytes of data into druid on a daily basis and this extensive data infrastructure support many products as i said and many use cases in this presentation i want to focus on two use cases the first one is what we call building target audiences so audience in our world is basically composition of attributes each device can be defined by different attributes that we can assign to the person who is using this device so for example the gender of this person the age of the person the interest of this person so these are attributes that we assign to a device all to a person and when talking about audiences we basically want to allow our clients to define a composition of this attribute so for example an audience can be all the females in a specific age with a specific interest so what you see in the screening is basically the audience definition screen our clients can drag from the bottom part from the bottom part of the screen the different attributes define the relation between them it can be an end relation oh and on all relation and our goal is to find the number of unique devices all unique persons and show the number in the blue box at the top the second use case that i want to discuss is what we call final analysis marketeers need the ability to measure their effectiveness of different uh funnels that they're running now in order to measure the effectiveness of a funnel they need to understand how many people reached each stage in the final so our goal is to analyze each stage and report the number of unique devices or unique persons that exist exist in each stage and this is how marketers again can understand the effectiveness of the of the of the different funnels and compare and choose the best funnels that work for them so i presented only two use cases but as you've seen we have diverse use cases and we've been looking for a single system that can help us solve um many uh different type of data problems we didn't want to have a dedicated system for each use case because then it means that our infrastructure will get very complex so uh we were looking for a single system now druid because the extensive query language the ability to fine-tune the system to configure it was a great fit for us and if you also think about it there is also a common thread to this problem essentially what we are trying to do no matter whether we are talking about building target audiences or analyzing a funnel we are trying to find a number of unique elements of unique devices or persons in some stream of events so this is essentially the count distinct problem and we try to do that in real time because there are users who are waiting uh on the screen for an answer so it cannot take minutes to return an answer and it cannot take even tens of seconds it's too much and of course that we need to do that at scale because we need to scan a lot of information in order to generate an answer now unfortunately if you try to uh uh solve the conditioning problem and do that uh in real time and at scale you will soon find out that there is no accurate space efficient and time efficient solution so the only relevant solution in our use case was to use what is called approximation algorithms now the idea band approximation algorithms is that you don't get the exact result you don't get the exact number of unique elements in a set in our case you get some estimation which should be close enough but in return for the loss of accuracy you get significant improvements both in the time required for the solution to run and in the space required for the solution so we chose statosketch which is an approximation algorithm that is based on the idea of kmv k minimum values where k is a constant that defines the number of samples you store again in approximation algorithms you don't need to store all the elements in your universe you only store a center and this is how you also control the space requirements of the solution and the minimum values refers to the process of choosing which elements you're going to store in your center with kmv and data sketch of course you can estimate set cardinality which means that you can get the number of unique elements in a set and for us the biggest benefit of teta sketch and kmp was its great support for set theoretic operations which is very important feature for us because it means that we could calculate intersection and union between two sets or more without losing too much accuracy and again this is an important feature because we are going back to the first use case building target audiences essentially what we are doing there is calculating an intersection over union between multiple attributes now the only downside of kmv and tata sketch was when we tried to intersect a very small set with a very big big set only in this specific scenario the error was relatively high and the tie will show with you some ideas on how you can mitigate this specific problem so we know that we want to use status sketch but where to find a concrete implementation of tetrasketch and luckily for us status sketch is supported in druid via the great data sketches library from yao and the way that it works is that during ingestion time based on your configuration the different sketches are being created and stored as druid segments and then during query time the sketches are being pulled from the segments aggregated based on the configuration and you get a final aggregated sketch that represents the number of unique elements in your set now this is a very basic explanation about data sketch and how it works in druid if you are looking to understand and learn more and maybe understand how it works behind the scenes you can watch these short videos that you created few years ago okay so we learned about the conditioning problem we understood that this is a very difficult problem to solve especially in real time and at scale and we understood that teta sketch can help us and the tata sketch is supported in druid but now we need to leverage all that knowledge and understand how to fit how to build the full pipeline and this is exactly what the thai is going to show you thank you thanks again okay so he uh told you about tether sketch and druid and what we're going to do in the next slides is essentially go through a data pipeline and focus on the main areas that you mentioned earlier so data modeling data ingestion retention deletion and query optimization so starting off with data modeling here's a made-up example of a data source so essentially a table in fluid uh you probably see it as you know the required timestamp column there's a dimension a string column which we call audience name which is basically the audience as you can explain that we identified on a specific device now i called it a naive approach because if we need to count the distinct the number of distinct devices we have for a given criteria for example how many tablets were used in a given day then we kind of have to do a not only a full table scan but an inefficient uh query that will take a lot of time and resources to complete now a better approach would be uh to replace this device id column with a teta sketch metric uh based on the uh tether sketch module detect you mentioned and then if you want to know how many tablet devices were used in a given date we only need to take this specific metric and extrapolate the approximated number of unique devices used now there is some kind of uh uh con to that right because if for example we need to intersect a very large set like tablet devices with a very small set like druid meters which unfortunately for us there is any help only handful then intersecting these two tata sketch metric will probably result in a very high air weight one way to mitigate that is to extract the device type into a new dimension and then if you want to know how many uh grid committers were using tablets for a given date there's no intersection there's only one specific data sketch metric that we need to take or extrapolate the approximated number of unique devices from another thing would be slowly changing dimensions so for example let's say we want to change the audience name from football fence to american football fans because we all know american football and football is not the same right and so with the nave approach if we actually store the audience name then changing uh the audience name in this example would require us to either re-ingest the data or find some other hacky way to do that however a better approach would be to leverage druid lookups lookups are essentially a mapping between some kind of arbitrary id like this audience id here and some kind of string like this audience name here now a lookup um can exist in a table and in a relational database in files or even in kafka and then the druid processes that are used for serving queries will load those uh mappings or that lookup into their memory and use it when serving corrects now our factory below essentially our data source would look this way after changing it so rather than storing the audience name we'll only store the audience id which is uh obviously more efficient and then we can use it to solve queries now you probably ask yourself that's that's great right but we're talking about serving uh customer facing dashboards and our customers don't know what is this arbitrary id and they don't care they care about the actual audience name and so the way to actually present the audience name is to either do a join or to use the lookup sql function and replace the audience id with the actual audience thing and finally when we want to achieve what we started off with meaning changing the audience name we'll only need to do it in one place and this is the origin of the lookup table so to summarize the data model section we talked about how you can use tetra sketch for fast and efficient account distinct and we told you you need to pay attention to intersections i showed you how you can leverage lookups to handle slow to handle slowly changing dimensions and we also just suggest that you check out the schema design page on druid's website now moving on to ingestion uh there are a few methods to ingest data into grid there's the streaming or the real-time ingestion you can do from kafka kinesis and you can also ingest it in batch either using the hadoop based or the native batch ingestion it might not be a surprise that we chose the adobe's ingestion because as you mentioned we started off with grid quite early on years ago now there are a few reasons why we chose that first technical considerations well when we started off with grid uh the adobe suggestion was the most mature option it's also very scalable so if we have a significant spike or a terminally alternatively a significant drop in the volume of our data at any given point in time we can either scale up or scale down a hadoop mapreduce cluster to accommodate the new volume of data the other thing was business requirements so we're talking about big data trends right so campaign trends and stuff like that so what we really care about are daily trends and so ingesting the data once a day or even a few times a day was good enough for us now the adobe's ingestion is still being used in large parts of the druid community and so we wanted to share a few tips with you the first one is many times you'd want to ingest data in parallel from different ingestion tasks from separate adult clusters to different data sources and so the way to achieve that is using affinity which is essentially a mapping between a middle manager and a data source and so it allows you to point your hadoop cluster to a specific middle manager and ingest to data source a and then to point a different adobe cluster to a different menu manager and to ingest to data source b in the same time another thing that comes up quite often is how i can mask sensitive properties such as credentials in ingestion trust logs and you can do that by using the druid startup logging mass properties now i still advise you to take a look at the injection task clocks in the first couple of times because some sensitive information may still be logged even if this is enabled now it's very important to understand that ingestion tasks actually run two mapred mapreduce jobs by default the first one is called determined partitions and basically what it does it goes through the data and based on the target holes per segment config property it determines how many shards or partitions or segments those are kind of used interchangeably should be produced from the ingestion task the second one is called index partitions and based on the number that was generated from the first job it will actually build those partitions or charts now in some cases for example if you know in advance uh how many data you're expecting or if the data volume is kind of fixed you can actually significantly speed up ingestion if rather than using target holes per segment you set num shells in advance in your partition spec and in that case it will completely skip the determined partitions and only run the index partitions now mapreduce jobs even with this uh aforementioned optimization can still be quite slow and so another way to optimize them is by pre-aggregating the data in an upstream process if this is applicable so for example what we did as i showed you earlier in our data pipeline we have a spark application that processes the data and so in that spark application we actually write tether sketch objects uh based on partial aggregation of data and then when we ingest the data into druid we set the is input data sketch config property to true so druid knows we're ingesting data sketch objects and not [Music] raw data so to summarize the ingestion section uh i showed you the multiple options that exist i explained why we chose the adobe ingestion and showed you a few ways to optimize uh your hadoop based ingestion including for example using num shards or pre-aggregating the data there's a very interesting working progress done by the community led by the awesome julian jaffe around being able to ingest data directly from spark into druid's deep storage and you can check out this git issue after we modeled and ingested the data it's very important to uh manage it and retain and delete it properly and let's see uh how so there are three terms you need to be familiar with the first term is load rules which determine which segments which segments should be loaded into that cluster meaning into the local disks of the historical nodes form the deep storage and this can be done by interval or period for example the last 30 days they also determine how many replicas per segment will be loaded the opposite of that are draw pools which determine when segments should be dropped from the clusters meaning deleted from the local disks but not deleted from deep storage and finally kill tasks are those tasks that permanently delete all information about the segment and completely remove it from deep storage now segments actually have versions so whenever you re-ingest the data or add new data a new version of the segment will be created now all versions take up space in deep storage however only the latest version is actually being used to serve queries and it's marked with used equals one in the metadata store and what we suggest you do is to specify your kill task interval as wide as possible don't be afraid to do that because it will only delete those segments that are marked as unused meaning used equal zero in the metadata store why is this so important let's take a real-world example we had one one of our data sources that had one retention period meaning we wanted to serve queries up to the last year our deep storage is aws s3 you can see that initially our killtask interval was everything ordered in 30 days meaning delete all the unused versions that are ordered in 30 days but again still maintain those versions that are being used for one year and so it took up over 350 terabytes in deep storage and it cost us over eight thousand dollars a month now by only changing the kiltus interval to delete all unused versions uh that are older than two days we managed to drop the used storage to only 15 terabytes and the cost per month to only 350 dollars so we are talking about over 95 percent move improvement for only one single data source another cool thing you can do is dimension based etl now druid as probably any other database out there allows you to do uh ttl based on some kind of time stem column like this time same column here but however consider this gdpr-like use case where in this made-up example we have data from u.s countries in new countries like sweden let's say you are allowed to store up to 90 days worth of data from us but only up to uh 30 days worth of data from new countries next week at the moment as you can see we have uh data older than 30 days for both countries now let's say we also want to add new data into the data source which essentially what the result we're looking uh to get is something like that right so to delete all the data ordered in 30 days for sweden keep data older than 30 days and up to 90 days for us and also add data for today for both countries and the way to achieve that is using the adobe's ingestion with multi-type input spec that allows you to combine other input specs so you can take your existing data in your data source and add or combine new data arriving for example for form s3 now in the type data source you can actually use a filter as part of the ingestion spec to filter in only the data you want to keep i know it was a very high level explanation but my colleague and i wrote a deep dive blog post about it and you can check it out in this link so to summarize the retention deletion section we talked about load and drop pools we talked about kill task as a way to permanently delete data and we talked about the cool dimension based ttl trick so after we model ingest and manage to retain and delete the data let's talk about queries so druid has two ways to create data either native phrase which are based on json query language or the read sql again no surprise we chose the native praise uh mainly because you know droid sequel didn't even exist when you started offer 3. however do its sequel is expanding rapidly with each version so for example the union all operator was added in druid 020 and the most recent druid 022 brought many features and improvements uh including the vectorized query engine support for data sketch aggregato which means data sketch queries got a significant performance boost that's why today if you start with read you'd want to start with grid sql now let's talk about query optimization and kind of the common thread in our slides or working with stata sketch so let's see how we can tune teta sketch size for performance remember mentioned tether sketch and kmv k is the size of the sample that we take from the incoming stream of events and this is basically the equivalent to the size that we will use for the data sketch when querying it so queries with relatively high size like 65 000 are heavier on resources that's they will take more time if you switch to a more moderate size like 4000 it can significantly improve your speed and this is very useful if your coils have no intersections so there will be relatively small effect on accuracy now the flip side of that is tuning that sketch for accuracy let's take this again real world example from one of our data sources you see i am using the druid sql language here i'm using the approximate count this distinct dstata function and i provide the name of the datastage metric the name of the data source and some kind of criteria now the result here was uh something a little over three and a half million right now by only slightly uh slightly changing it and providing a second argument as 65 000 to this approximate count distinct theta function we got a result that's around three thousand higher and i can assure you it's more accurate and why is that because this approximate count distinct theta function has an optional second argument of size which defaults to 16 000 and so without providing anything we got a res we got a result that's close enough uh and probably faster but a bit less accurate than we when we provided a higher size thus we've got a more accurate result so to summarize the query section we talked about the two query methods why we chose native phrase but why you'd probably want to go with grid sql we showed you how you can tune teta sketch to balance between performance and accuracy and also going back to the same uh working progress done by the community it eventually it will also allow you to create data directly from doing steep storage using spark okay so to summarize essentially what we showed you is how you can ingest terabytes of data use it for customer facing dashboards and do all that in a cost efficient manner just before we wrap things up a few things we care about so first of all women in big data is a worldwide program that aims to inspire connect grow and champion the success of women in the big data analytics field there are over 17 000 members worldwide and everyone can join the goddess of gender so we really encourage you to find the chapter near you using this women in big data website there are a couple of past talks that we gave including this druid intro session uh from big data london 2019 detective mentioned and uh quite recent final analysis with apache spark android talk that my colleague and i gave at the latest data in ai summit there are also a couple of blog posts that we mentioned including the data retention deletion in apache druid and the second part of that which is more focused on the deletion i would also like to mention that druid summit is uh coming up in november 9th and 10th and also uh druid celebrated its ninth open source anniversary earlier this week and you can see the introducing druid post android website by eric checher that announced originally announced open source android back in 2012 that's it for us we really appreciate your time and you participating and attending this talk today and we will also uh really like to get your feedback so really feel free to reach out to either your kids or myself via twitter linkedin uh and really provide your feedback insights or even follow up questions thanks everyone okay so today