BDSBTB 2015: Anirudh Todi, TSAR (the TimeSeries AggregatoR) -- Billions of Daily Events in Real Time
thank you um so my name is unnie as I said like I'm a graduate from UC Berkeley I did my undergrad there twitter is my first job out of college and one of the main things that I've worked on on Twitter so far is a technology called SAR SAR stands for the words time series aggregator so let's just dive into it and I'll try and leave time for questions at the end so at the very outset like what is SAR so sorry is a framework and a service infrastructure that allows one to perform time series aggregation so what is what is the word time series aggregation mean if people in the room are familiar with something have used Google Analytics it like if you have your own website and you embed some Google Analytics code in their Google Analytics will provide you with a dashboard that shows you how many people have visited your website day / day and it gives you like a time series of visits to your website it can break this down by various like segmentation like the location of the different users the gender of the different users so it gives you some metric over which you are aggregating and then it produces some something that you care about like so in your case it might be visits to your websites and then it's producing like a count in this case so in the case of Twitter time series aggregation is a pretty common problem we anticipate that it probably solves like about sixty seventy percent of the various analytics needs that we have and previously we had various engineers within the company who would build their own analytics frameworks that performed and repeated the same problem and so we abstracted away all of this into one common framework that has a lot of optimizations built in and that allows it to scale to the kind of events the scale that like Twitter needs it essentially so as we said time series aggregation a twitter is a pretty common problem some of the places where we would use it are various data products so I'm an advertiser my I'm BuzzFeed or I'm nike and I'm advertising on Twitter I want to see how my promoted to beat campaign is performing how many people are clicking on my promoter tweet / day 1 versus day 2 versus day 3 how many people are favoriting my Promoted Tweet how many people are following me so there's some good surfacing some metrics in the time series to advertisers it's very useful internally for different business metrics for example if I'm working on the platform relations team at Twitter I want to know which client applications are using twitter's api the most is there a particular API endpoint that is most common that is the number of calls to this API is going up over time is there a particular client app that is getting a huge number of tokens being granted because there's a huge amount of adoption is someone not using our API in the best way so there's a lot of internal business metrics that are useful for us to track and those also fall into the class of time series aggregations and then it's also useful for engineers themselves they managed various services so they want to track their service health over time what is my success rate over time how many requests is my service getting over time how many failures does it have over time and so on and so forth so there can be many classes like externally facing internal business metrics in internal use that all fall into the category of time series aggregation so the parts that make it hard to do as always is scaling such a system Twitter processes tens of billions of events per day we want to it inherently is a real-time platform so you need to perform real-time computation and you need this to be a robust system you can't really have your time series aggregation job go down and then tell your advertiser hey like we don't like have metrics for these 15 minutes because our aggregation service went down because it got too many events and it couldn't scale here knees so how do you make it robust while giving real-time metrics so time series aggregation like solving this problem wasn't solved like over time it wasn't solved just by me when like I joined Twitter it's been a multi-year effort over many many years you many people in this room I'm sure are familiar with MapReduce a Twitter we built out a library called scalding that provides an abstraction it's written in scarlet provides and it's available on github it provides an abstraction over MapReduce and allows one to specify MapReduce jobs easily we had acquired a company called back type that a technology called storm that provide real-time computation and we've been working and so back type so storm would provide real-time computation scalding would perform offline computation we then developed a library called summing but that is also open source and is available on github that combines scalding and storm they provide the where you can plug in any online platform and offline platform and it gracefully allows you to write one summing good job that translates that gets translated into an offline job as well as an online job and we will discuss what the use cases of this are so the summing but library is kind of at the very core of the SAR framework but it's a library and what's our provides is provides a framework it provides a service infrastructure on top of summing word it provides the tooling in an easy manner so you can write an offline job to read in raw logs specify dimensions that you want to aggregate on compute the metrics that you care about store it and data store in an efficient data model provides a query service by which you can query the data it makes it easy so that you can aggregate data on a monthly granularity or an hourly granularity or a daily granularity it makes it easy to perform Restatement tasks like that fills so there's a lot of service infrastructure that is provided by the SAR framework as it builds and expands on top of the summing but library so it's been a multi-year effort over over the course of many years is it said to develop the SAR framework um and the whole goal is that it abstracts away all the all of the business slot or all of the data pipeline logic from the user and allows the user to only focus on their business logic so a typical aggregation job like what does it look like somewhere you have draw event logs that exist on typically at in the case of Twitter on HDFS what do you mean by an event log & Event log is say I go and I favorite Sally's tweet that would send an event back to Twitter saying that user honey went in favorited Sally's tweet at a particular time stamp in what data center did that event take place like what client application ID did I use what OS did I use and so on and so forth so there's one log event that encapsulate seeeeee a large number of feel like maybe about hundreds and hundreds of field hundreds of bits of information about that event this gets sent back to twitter data center and then using the scribe pipeline this these events get logged on to HDFS your aggregation job typically only it cares about a few of these fields like if I want to just aggregate a number of favorites from each to it then all i care about from this log event is whose what was the tweet ID on which the favorite action occurred that's the only event that I actually care about from the hundreds of feels that there are so there's an extract phase where I extract the fuels that I care about from just that log event then I perform a grouping which is the aggregation the dimension that I'm aggregating on which in this case would be the tweet ID so I'm grouping on the tweet ID and then I perform some measurement which is a count in this case I'm summing so I'm summing across all the favorite events on that grouping dimension that is the tweet ID and then I stored this aggregate somewhere which could either be in a relational database so you can access it or in a key value data store it depends on your scale and from your applications use case and we'll talk about some of the use cases that there are so to go through an example again like the event logs could look like I have some API endpoint I have some response code of the event that took place on when and when this when I hit this API endpoint there's the client application like the or the OS type whether it was an iphone vs and web versus android etc that made the actual you call i extract only the feel that i care about so in my previous event log as i said we could have hundreds and hundreds of fields i extract only the couple of fields that I care about in my aggregation job I perform some grouping which in this case here I'm grouping by the API endpoints I'm computing the number of I'm computing the count of like they're different types of response codes based on each API endpoint so I'm grouping on the API endpoint in this case so I essentially get a map and then I collapse my map down so it gives me the total number of calls that was made to each API endpoint so I'm performing a summation over here so here's what I'm performing the measurement and then I can store the data somewhere typically the two types of places where you would store the data would be either like a key value database useful if you want to store huge amounts of data that you want to access with low latency or you might store it in a relational database typically you can store as much data but it allows you to perform complex queries you can do sorts group buys limit queries etc which you can do and say a key value database so going through a couple of more example just to drive home the point before we dive into some of the more details about house are is implemented if you look at the example of API aggregates which means that multiple calls are made by different line applications to twitter's api i want to bucket by each API call and I want to group on some dimensions which might include the endpoint that the call was made in the data center or the client application ID that was used to make the call to that particular API I might want to compute as a person at where I might want to compute various metrics which might include the total number of calls made to that API endpoint the number of unique users that access that particular endpoint the mean response time for that particular endpoint and so on and I might want to write this output out into vertica because the key space in this case isn't that large number of endpoints twitter has maybe hundreds of thousands it's not in the millions of the billions so the key space isn't very large so i can write this data out into vertical also a data with like this is typically not going to be surfaced externally outside of Twitter it's going to be used to build internal facing dashboards so the latency requirements on it are less so i can write it out into a relational database like my sequel of vertical and it would scale to the needs of an internal dashboard even if it loads in a few seconds of time it's probably fine another example similar that requires some form of time series aggregation but that is a little different from the previous one is that like if I want to open my iphone app that is Twitter at the Twitter app on my iPhone and next to every single tweet I want to see the total number of unique users who have seen this tweet now this becomes a product facing feature that is going to be accessed by millions of users across the world so the needs are very different the previous one was an internal only facing feature here I have an external facing product so the requirements are different it needs to be much lower latency in each to it's a much higher key space because I'm aggregating on every single tweet in this case so the key spaces millions maybe even billions in this case but the essential core of the problem is the same where I want a group or bucket by some dimension which in this case would be the tweet ID I want to produce some metric that is I want to count something which in this case is the impress on this tweet ID and then I want to store the data where previously was able to store the data inside vertical here and I might want to store the data inside a key value database key value store so that you can access the data and low latency and you can act like it can scale to the needs of the Twitter app essentially um and then since it's written out into a key value store I need some way for the Twitter app to actually obtain this data so I expose the theta Y a high a select query service which typically in the case of filler speaks thrift and you might potentially want to write like a sample of this data to vertical to perform like validation or perform internal metrics like tweets that are sent out by people who have more than hundred thousand followers like impressions on those tweets maybe I write that out to do what it goes just a small subset of data so i can like inspect it run some queries if i want to do some debugging some quick debugging but it's hard to do that when you write out to a key value database these two problems slightly different but a kind of the code they're both Prime series aggregation problems that have different requirements based on their application use case so what is some of the basic problems that come up when you perform time series aggregation the biggest one is service interruption twitter is a real-time platform we want to provide for all time and let x to nike so that if Nike send out a Promoted Tweet within a couple of minutes they can see how many people are impressing their tweet so this has requires some real time component to your analytics job what happens when your real time component suddenly goes into like a garbage collection issue it gets too many events in a single a Promoted Tweet and so it suffers from data skew and is unable to keep up some node goes down and hits a bad host for whatever reason and so it loses a few minutes worth of data it's unacceptable for Twitter to display data to an advertiser and say hey like from like 1155 to 12 we don't have data because our all-time topology went down we need to be able to build in some fault tolerance so we can surface theta even for that period to the advertiser how do you build in that particular fault tolerance in your analytic system it requires performing do this data store coordination so analytics data could be used for various reasons you might want to store your data inside a cache so you can retrieve it quickly you might want to store your data on disk so that you can retrieve data from years by because very often you want to compare over your comparisons on how like you perform during say the Super Bowl or during new years or so on you might want to store your data back on HDFS so that a further downstream analytics job can pick up the output of your aggregates and then do some further processing you might want to store it inside a relational database you might want to store it inside a key value store so there can be many kind of places where you might want to store it and you would have to define a gator model for each one of these further very often you when you first start off working on analytics job for say an application or a use case you build like some version like a V one of your job you get user feedback which could either be internal customers or external customers and then you iterate on this you change your schema the your product manager says hey like you've been providing impressions on tweets only let's now provide impressions on tweets also broken down by client application ID so you want to evolve your schema and how do you make your schema and your analytics pipeline in your job be flexible to this kind of schema change so it can evolve very easily um that becomes a challenge and then lastly and probably one of the most important things that and that gets overlooked very easily is how do you make it easy to perform back filter restatements because as I talked about the example where as you evolve your job into a v2 version your PM or you're like users would ideally like to see that same evolved version of your job for say the last three months during which v1 was there as well so how do you go back and like update this historical data that you've already computed or say I'm an engineer engineers make mistakes it's human error like I made a change to a job that's running in production and I deployed it with a bug and so I've been computing some incorrect metrics or incorrect data for say a couple of hours and then I realize this and I fix it and I deploy the correct version of my job but it means that for those two hours in the middle I have stored incorrect data in my data store how do I go back and update this or repair this historical data that I've computed so making it easy to perform back fields become the challenge it's very very common for and I've seen this many times and we had this problem for the first six or seven months before we actually decided to build start is that we would have human error and then it would become a real pain to actually update or he should update this historical data so making backfill the first class citizen was one of the primary goals as we set out to like develops are essentially um and the goal is some of these are some of the problems that I've outlined and the goal whereas building outs are was to solve these problems in a general way so solve them once and for all so sorry's like main design principles that allow us to solve some of these problems I'll walk through a few of them the first one is hybrid computation so hybrid computation means that every event that comes into Twitter gets processed twice by us this means that like any vented I said was say i favorited Sally Ann's to eat that's an event that event gets sent into Twitter it gets processed by some service that service sends that event to our real-time topologies so that they can pick it up event up in real time that event also gets scribed on to HDFS so it can be picked up by an offline job so here is where the core of summing word comes in when I said thus our framework is tied very closely to summing board is we have an offline job that processes that favorite event and online job that process of that favorite event we then have the query service that is extremely smart and what the query service does is it favors the data from the offline whenever possible and only for the period the online data offline job has not run does it actually query the real time layers so let's look at this example it might help illustrate what I'm talking about an event comes in to Twitter that's a it would go into the batch clear as well as the speed layer so it goes into both clears get processed by both of those aggregation pipelines and then the query service would first query offline and it says hey for this tweet idn for this hour do I have a count that has been computed by my offline job offline jobs typically run at say a cycle of say every four hours every eight hours a daily job or so on but i want to give up-to-date counter to the advertiser so I favor the results by the offline and only for the period that the offline job has not run up to the current time do I take the data from the real time component so what does this mean this means that offline jobs are typically more stable they are like MapReduce jobs so they can process much more amount of data so it gives me the stability of an offline layer while giving me the recent real time so now say if my real time layer like goes down or goes into GC I'm confident that in a few hours whenever my offline job comes along it will have processed that log event and store that log in and would compute the aggregates that are required and actually surface the real number over to advertisers so it is eventually fault-tolerant this system becomes eventually folder and that's like the key aspect of providing hybrid computation the second the second design principle is we separate event production from event aggregation what I mean by this is events coming together like a favorite event comes into Twitter as I said like in the in the example that we talked about where I have multiple log events I extract the fields I care about I perform some grouping then I perform some measurement on the field that I have grouped on and then I store the data somewhere these aspects are common to every single job the parts that differ from job to job are what dimension should I group on what are the metrics should I that I should produce what granularity should aggregate or should it be hourly or daily or minute Lee what datastore should I store the data end but those can all be configurable values so the actual business logic of taking the timestamp of the event rounding it down to say the our boundary or figuring out what the data model is that should store the data inside my key value data store those are all handled by the SAR framework the only thus our framework what the user needs to do is specify how he should produce the events that is I have a log event that i'm reading from what are the fields that I care about and what are the dimensions that I want to produce the further downstream steps of actually writing the business logic for computing for providing the groupings for creating your data model for spinning up your query service for spinning up your observability your alerts etc that is all handled by the SAR framework the the whole point of this the event production and event aggregation difference the distinction is to allow the users to only focus on their business logic and all the other workflows are handled by the SAR framework this point will become a lot clearer once we actually go through an example of a SAR job which is coming up in a few slides um the third design principle is a unified data schema as I said like um an analytics pipeline would want to store data say on HDFS for further downstream analytics jobs to pick up it might want to store the data inside a key value data store in our case Twitter calls it Manhattan which is like a custom key value store people don't think of it something like a Cassandra you might want to store it inside relational database like vertigo my sequel or in a cache further we might want this to be extensible to other schemas like today if like we don't support HBase but if we want to store it inside HBase we can like easily plug in a new sink for HBase and then every single customer or Sark and then like write data out into HBase and so on so how do you provide a unified data schema that scales to the needs of billions of events and that is not that doesn't take up too much space storage but that doesn't take too much time to query or right into and there is also sexibl to schema change that becomes a challenge that this our framework tries to solve so typically that Gator schema coordination is I have data coming in from either like a Kafka spout which is consumed in real time or some event logs in HDFS I produce some aggregates that then get stored inside vertical or inside like Manhattan for example and then the fourth one is that we provide an integrated service tool kit and so what this means is any analytics pipeline at twitter has to be robust so you need to have observed lydian alerting on your real time jobs you need to have observed lady and alerting on your offline jobs on your query service you want to have some form of anomaly detection so like if like your computing data and your data that you're computing is corrupt you want to be aware of it so that you can surface that that there's an anomaly to the advertiser so their DD integrity checks there's alerting an observed lady there's ease of like there are so many different moving parts to any analytics pipeline so providing an easy tool kit so that you can easily manage them you can deploy them you can make changes to them that was one of the goals are to reduce the operational burden and load on the user so I've spoken a lot about some of the problems of time series aggregation some of the design principles behind SAR let's actually maybe dive into a specific SAR job and that might help like just fit motivate more of this more of this discussion so the example that I wanted to go through was the example of tweet impressions we talked about this we're like say if i open my twitter app and i want to count impressions on every single tweet ID so here I'm producing I'm grouping on the tweet ID and producing a count since it's a user facing product I would want to store it in I side a key value database the challenges here are we get 500 million tweets in a single day we get tens of billions of impressions because any tweet is found out to all the users followers who can all impress the tweet impress meaning view the tweet I want to get real-time updates and I want this whole system to be production ready and robust so how do I build such an analytic system so a minimal side project has three components to it we have a scholar SAR job so this our job is written inside scholar is it built that on top of something word which is also in scholar there's a config file and there's a thrift IDL where you specify the dimensions that you care about so there are typically three components and we'll walk through all of these three components now so the thrift IDL is where you would specify the dimensions that you care about in this case I'm just grouping on the tweet ID like I'm computing impressions on a tweet ID so the only dimension that actually care about is the tweet ID so i define it thrift struct and call it just have one field inside it here is the actual SAR job and we'll walk through this more so this is where we'll spend some of our time so sorry the parts that are in blue is the custom SAR dsl that we built out which we believe makes it easy to specify what you want to aggregate on so the on keys block is where you specify the dimensions that you wish to groupon in this case i want to group on the tweet ID so you would write tweet ID over here because tweet ID was part of your thrift struct we pick it up using screws reflection over here inside the on keys block you want to produce some metric which in this case would be account you want to produce a count of number of people who have impressed this to eat ivy other metrics that you could have here might be a unique computation or ranking a standard deviation mean you could have different kinds of metrics that all ship out of the box within the SAR framework then you have a sink to block which is what data stored you want to write your data out to do since this is a user facing product we write our data out into manhattan manhattan meaning or key value database in this case or i might want to write my dear doubted to vertical in which case i would replace this sink to block with verdict instead and then the front producer block is actually what calls out into the summing buddy API so this is at the actual summing word fragment which describes the event production logic so if you walk through this this is saying read from the client event source the client event source is a stream of events that are sent by every single Twitter client meaning twitter for iphone Twitter for blackberry twitter for ipad twitter for android they all send um a struct called the client event struct that is sent by clients back into twitter so i want to read from these events that are sent by all of the different clients that are coming in in real time or in offline y HDFS now every event is getting sense of favorite events retweet events reply events I in my description of my application I only care about impression events so I first filter down to impression events by calling as is impression event library or helper function and then I produce the event where the event that I'm producing is I only extract the tweet IDs I'm calling event or tweet ID so the event could have multiple in lots of information every user that perform the impression event on whose tweet it was that the impression he wonders perform what was the timestamp what was the data center so on and so forth I only care about the tweet ID field which is the only feel that I extract and then I output a tuple which is the time standard the event occurred and the fact that this my fridge truck that I am aggregating on that I just defined in the previous slide so this is where I'm producing the event notice that in this slide there is absolutely no aggregation logic being specified I have not defined how I'm going to sterilize this ski that I'm producing I have not defined how I'm going to round this time stamp down to an hourly granularity or a daily granularity or so on I have not defined how I'm going to produce my count ventric so where is this some happening where is the dot group or dot some that is happening none of that has been defined in mice our job that is all completely abstracted away from the end user by the sword framework so the surface goal is a user rights simply specifies these are the dimensions of an aggregate on these are the metrics I want to produce here is what I want to store the data and these are the events I want to produce the logic of performing the aggregation is completely handled by the Tsar framework the Tsar framework has enough information from this file to extract all the other relevant bits so the configuration file which is the last piece of the puzzle is over here so walking through it you can namespace your job so that different people and your team can have like a namespace you give your job some name which might be tweets or tweet impressions in this case the user who's deploying the job and origin which is when you want the job to start running from which might be say the 15th of April of 15th of May of 2014 so you have some history and then in the outputs block is where you specify where you wonder what granularity you want to aggregate on so I want to aggregate and write data out into Manhattan and say hourly granularity which is why I specifies a sink thought Manhattan and width of flakes a one-day or daily granularity in this case so I'm producing dating aggregates in this case so that's all the information that the user needs to provide so in our side job what are some of the things that we have specified we specified what is the event schema which has been specified and thrift how we're going to produce the events again I make a distinction reproducing events and aggregating events we specify how to produce events and we specified various like almost like config options dimensions to aggregate on time granularities you want to aggregate on the sinks you want to write to what we don't specify is how are you going to represent your aggregated data how are you going to represent your schema inside my sequel or Manhattan how are you going to perform the aggregation how are you going to connect to your underlying services that information is all handled by SAR over here and the beauty of the SAR framework is it provides you an end-to-end service infrastructure with a single command we provide we give you a command line tool that you can use and when you run the command line tool which is called the SAR command line tool you ants are deployed to the production environment it does a whole host of things for you it will deploy your Hadoop jobs for you your scalding jobs for you it will deploy and launch your storm jobs for you it will launch your thrift query service for you if you're storing data inside vertica my sequel it can also launch loader processes we can pick up the aggregated data stored by or offline on HDFS and storage inside Vertica it we use Aurora and my sources are schedulers for our offline jobs which requires some configs to be generated it infers those and Otto generates those configs it generates alerts for your bad jobs your query service job here for your online jobs a general observed lady for all of the different components of your pipeline if you've integrated with the anomaly detection framework of our thing it will automatically generate anomaly detection alerts for you so say your computing impressions on tweets and the total impressions on a particular day is say like some number say 100 x and on a particular day the total impressions ends up being 50 x like hey where something is off maybe the iOS developer made a mistake and like stop logging impressions on the iOS client so you get alerted to these kind of these kind of anomalies um if you're storing data inside vertical my sequel some table needs to be created some view needs to be created over those tables some projections integrated it will automatically figure out that I need to go into vertical create a table what is the schema of the table can be defined by the dsl so you know that you're aggregating on the tweet ID and producing counts so you know that you just have to have two fields in two columns in your table which is the tweet ID and the count and maybe their day granularity so you have a day timestamp as well so all of this information and all of these various steps are performed by the SAR tool in this case so sorry builds on top of summing bud and makes it very easy for the user to build upon the frame to like do things essentially to just focus on the business logic and get an entire analytics pipeline up and running very easily so a bird's eye view of this our pipeline looks something like this where we have log data on HDFS typically which that reaches there by scribe we have real-time streams coming in which could either be say Kafka orchestral the log data gets aggregated on by scalding that is bad jobs the real time data or Kafka streams can irrigate it by say storm jobs these this layer is like what we call the summing bird lair so summing but job translates into scalding and storm then if you want to write data into vertical my sequel there's no like real time component we don't have a real time job hammering into Vertica but you can write offline data there to do ad-hoc investigations that can then be used for data exploration of our internal dashboards if you want the user facing product you could write your data out into a key value data store so you're offline an online job would go into manhattan using the summing bird storehouse batch libraries it took there's a thrift query service that queries this data and n can be queried by a public-facing dashboard to provide like a time swedish to the advertiser so that's roughly what a bird's-eye view of the pipeline looks like um so now say like what is the beauty of sorry framework is ok now I've built my analytics piping and I'm very happy and like the developers gone home but now product managers which are the bane of my existence suddenly decide to come in it's like hey like I want to do something different so the product manager is like okay I broke down impressions by tweet ID now let's break down and precious by tweet ID and also client application ID because that's more useful information that like I want a surface or gain an understanding into in a typical case you would have to go update your offline job update your online job update your query service update your schema inside vertical to add a client application ID column do a bunch of things update your observability your alerts etc etc in sorry sorry I skipped it and sorry you add one line into your own keys block saying I want to aggregate on tweet ID and client application ID and then in your produce events place you extract the client application ID from the raw event and you run SAR deploy and it will automatically do everything for you so it allows you very easy schema evolution in this case what about ok so my product manager says ok great you've done this work for me but your old version of your job has been running for three months i also want this broken on my client application for the last three months ok i run a command that says salt bad fell and I pass it a start time and end time and it automatically spins up a backfill job that runs in parallel to the production job so the production job continues to run onward and a backfill job runs in parallel to the production job and comes in and generates my new data for me this can be useful for repairing historical data as engineers make mistakes as we talked about as well ok now say I've been advocating on daily aggregates I now want to produce an all-time aggregate so how many impressions has the street gotten in all time or I want to produce hourly aggregates I if you remember my configuration file had as output block which set the sink and the granularity I now add a single line to my granularity which says width equals all time and I again run the salt apply to and it would begin to generate all time aggregates for me as well going forward see I have been computing total counts now I want to actually compute number of unique users so for example in the oil example say I have viewed a tweet at 11am and then I view it again at 4pm that would count as a count of two it's raw counts actually say I want to compute like reach like a freak like the reach and frequency like meaning how many unique users have seen my tweet so i can add a column that says unique user IDs and that now ni deployed and this will now begin and I extract the user ID field in there so now I now begin to get number of unique users who have seen my treat anyone see any problems they still say like I have number of unique users who saw my tweet on day one is say hundred number of unique you saw my tweet on day two is say fifty number of you need to do saw my tweet on day three is safe 75 so and I get a query to the southbury service that says give me the number of unique users who saw my tweet between day one and day three um anyone know how I would do this or any problem we're doing this sure okay how would you do that though okay anyone wanna try yep exactly so precisely someone said that their approximate counting mechanism so the problem is that if I have three unique guns I can't add the three to get another unique count so unique uniques can be backed by something called a hyper log log which is a data structure that can provide approximate unique counts in a very efficient manner so I would store a hyper log log in my data store that represents the unique count for day one similarly a hybrid loglevel that represents the unique count for day two ipod logo that represents the unique count for day three so when i get a query over the three days I extract the three hyper log logs I merge them together to form one big fat hyper log log and then I derived this hyper log log to actually give a unique count so this is all done at query time now I'm an application developer I don't really want to go and implement hyper log logs I don't want to do this kind of optimizations so thus our framework automatically figures out that hey I'm a unique I wanted a unique computation okay let's go and store hyper log blocks similarly if I want to do percentiles it goes and stores q trees which is another data structure that performs does percentiles say I want to compute top men things like rankings it goes and stores a sketch map instead so it has multiple derived metrics that are backed by probabilistic data structures that do approximate counting and these probability structures are even smarter so say for unique counts say like I have a tweet that only receives two impressions since it's probabilistic I don't want to report like three or four in this case I actually want to report to so the unique counts are exact up to a cut up to a certain cardinality so say I can set a car Noddy saying up to say thousand 24 I will maintain an exact set and only when the cardinality of my set exceeds this value will I convert it into a hybrid log dog data structure so it's approximate its approximate but exact up to a certain threshold and only when the threshold is like exceeds some amount after which you can begin to be a little bit probabilistic and these are all baked into the Tsar framework and the libraries that it uses in this case algebra um if it makes it easy for users to do this and then say I've been persisting data to Manhattan I want to persist data to my sequel as well I add a new line to my storage of that says okay sync to my sequel and then I run so our deploy again and now begin to have aggregates inside my sequel so a typical start workflow is I create a job i deploy it if i want to make modify my job i can modify it and then i deploy it again and if i want i can optionally run a backfill as well in this case so i have actually a few slides on SAR optimizations but the conference organizers tell me that i'm coming out of time so i know that the slides have been sent to a conference organizers and they are available online on SlideShare if he googled my name a lot of these optimizations and everything so I'll skip over these I'm happy to chat with anyone offline about them so they're a bunch of optimization that we've done to makes our like more efficient and scale but in conclusion I wanted to say that like the kind of three problems thats all try to solve is the computational management aspect of things which is I specify aggregation dimensions metrics and pine granularities sard goes on automatically figures out hi one of the uniques how I want to do rankings how I want to do percentiles how I'm going to like round data down to our league and let you go daily grind how am I going to take into account different time zones of advertisers it allows for dataset management so how do you store data in different schemas how do you evolve these data schemers how do you coordinate data my aggression backfills recovery and then the service management aspect of it or operational load the stardrive should reduce the operational load and people on users in general there are lots of moving parts how do you make it easy to manage these things how do you make it easy to observe these things provide alerts on these things lead to regression checks and various things sard gives you all of the above essentially does one key takeaway from my presentation it's the entire end-to-end management of the data pipeline is like SARS key feature the user only concentrates on the business logic um thank you if you have any questions you can reach out to me and twitter my handle is on you wrote OD or you can email me ani at twitter com we're also hiring so I recently became a manager of the revenue analytics team we're directly using the SAR frameworks if interested in learning more reach out to me after the presentation but thank you we have time for a question or two hello so you talked a lot about schemas being utilized in SAR sure I was wondering about kind of getting the schemas into sorry if you go back to that slide in the front producer event it looked like there was a method sure is impression event and UX trap values from the schema I just wonder a little bit about what that storage and kind of validation or how that information kind of flows through so that you know to say hey pull out the event tweet ID or know that it is an impression event uh sorry so I guess my question is mean that again what do the schemas look like before SAR how is that handle oh sure I mean the schema like the events that come into SAR look something like so the question was what is the scheme of events look like before they come into sir so any typical event that comes into SAR would be something thats a thrift encoded like typically thrift is what we use a Twitter so like a client event as we talked about might be like in this case it's not exactly a client event but think of it like is some thrift event which is say a flat structure or it doesn't have to be flat it could be nested is what are different fields in this case like the API endpoint a response code the client in the case of client event the different fields you might have might be the user that created the event the tweet on which the event was created what time the event was created which enter the event was created in whose tweet was being favorited and so on so those are the kind of events that come into SAR thus our job would now like pull out the feels that it cares about so all right trying to jump around here so in this our job like in the second to last line we pull out the fields that I care abouts in this case I pull out just the tweet ID fields by calling event we tidy and this is the thing that I send down now thus our framework Ben takes over and realizes that I'm aggregating on tweet ID and I'm producing counts so I should like do a dot groupon tweet ID and then do it like a dot some or something along those lines and then like it then see realises it using like a codec and stores it in binary format inside a datastore that can then be deserialized every time and there are various optimizations in the sour optimization slides that I had to skip over on how we store it efficiently so that it can be stored efficiently and queried efficiently as well maybe one more quick question I great talk there by the way so I have a question about open sourcing some ability is actually dependent on the underline of scalding and stone you actually release the paper about the ear one which is going to be three times faster so you're actually using storm we're using arrow no production and when are you actually but the plan about open sourcing Thailand you won sure so storm is I believe it's open source we have been developing a new computer aim work at Twitter in general which i believe is not open source but there are talks of like open sourcing that sorry is something that we've talked about open sourcing though there aren't any plans to do so at the moment the main the two main reasons for doing that is one it's a lot harder to open source of framework then it is to open source a library the amount of work required to do so as much much greater and the second reason is just that sard is very very closely tied to a large part of Twitter's infrastructure so like for example one of the big goals that begs are provides is how do you reduce the operational were clear in the user because it makes it easy to plug into observability systems our loading systems because the start to like integrates with them already like it gives you that to free for any application developer a Twitter but an open source that's not exactly going to work so the parts are started that are useful in some ways to open source would be say this our dsl and how it like fans out into different like probabilistic data structures that are in the background that's something that we've considered it's something that just hasn't been done or prioritized as yet because it's again it's a lot harder it's all very tightly coupled with Twitter's infrastructure at the moment thank you re ok thank you