Twitter Heron in Practice

data.bythebay.io:Karthik Ramasamy, Twitter Heron in Practice

Recording: data.bythebay.io:Karthik Ramasamy, Twitter Heron in Practice

[Applause] thank you nice to be here and uh so the title of my talk will be Twitter Heron in practice uh how many of you know about Twitter heren in the first place okay not many looks like considering the fact that the paper has been downloaded uh close to 18,500 times wow so okay so I'll be giving a kind of a little bit overview of what Twitter heren does and uh so feel free to ask me any questions that you might have because uh I have a lot of details in my head but uh unless you ask me question then only the details will come out you know so anyway so with that short introduction I will be delving into the talk so the talk outline goes as follows and initially I will give you an overview about what heris and uh we kind of uh introducing the notion of what we call as a micro streaming engines and uh then I will discuss in detail a couple of problems that we face when Twitter heren is running at a large scale uh for the last two two and a half years and uh then I will delve more detail into what are the resource conception of Heron itself and finally I'll do the conclusions so as you all know Twitter is all about real time and we do a lot of realtime process Ing and that includes Computing realtime Trends based on uh analyzing the tweets what is the emerging news that is coming out and also there a lot of real-time conversations millions and millions of real-time conversation flows through those tweets and uh in order to identify and bubble up those conversation we need to employ real-time processing and also we do real-time recommendations when uh we inject an ad uh we inject into the stream depending upon the context and those contexts are continuously computed and that is injected in real time and finally we also have realtime search where the tweets have to be indexed the moment it arrives within a uh 100 milliseconds so that the results of the search can include the newly or the fresh tweets as well so that it feels very close to real time so so which means we are collecting billions and billions of events and how to process these events in order to provide all these various different real-time processing uh is a big challenge for us and uh initially as you all know in 2011 the open source storm and storm was one of our uh main stay for the from 2011 to 2013 then uh we faced a lot of issues with the storm I would suggest like I'm not going to reiterate all the problems that you faced with respect to storm but I would suggest to read the paper that we I will point out to the paper and please read about it if you wanted to know more about it and uh so I'm going to jump directly into her itself so so just to give a overv of heren I will introduce the heren terminology and heren terminology we just borrowed from storm and to keep it simple uh so it's the the notion of a realtime job is called a topology and a topology has a is nothing but an a dag and the vertices of the D represents the computation and the edges of data sorry edges between the vertices are essentially streams of data topples and there are two type of vertices spouts and spouts essentially like uh what they do is the tap into the data source and extract the data and inject the data into the job itself and some of the examples includes Kafka castral and how many of you know that Twitter has open source their uh new queuing system or messaging system called distributor log okay there you go so it's pretty it's pretty uh very good has been in production like 60 or 70% of our topologies are running in uh uh distributor log now and uh so it doesn't mean that we are to stick with only messaging systems but also with myql and post and a very few others as well the second one is bows the second type of vertices is called bows and Bows are essentially represents computational elements where you process incoming tles and emit outgoing tles depending upon how the processing goes and some of the examples include uh filtering aggregation join or any arbitary functions so how does a heron topology looks like like you have a couple of spouts uh feeding into the next stage bows which are like bolt one bolt two and bolt three and they are turn feed to the next stage of bolts bolt four and Bol five and uh one of the biggest differen is like compared to map Ru is the fact that uh there's no limit to the number of stages that you can stitch to you can stitch as many stages as you want we have jobs internally that is running like more than around 10 stages or 15 stages that's common so the word count to give a concrete example of how a topology looks like uh let's take the example of a word count where you are counting the uh uh stream of words that are coming out of a tweets so in order to accomplish this you need a a tweet spout which in turn Taps into the Twitter feed and uh that tweet spout will pass it the Tweet entire tweet into what you call a p tweet Bowl which is uh responsible for breaking the Tweet into a smaller set of words and finally you have a word count bow which in turn counts the distinct words that is AC occurring across the tweets so that you can spit out the count in a uh periodic fashion and uh from a database power length we call this as a dag or something called a logical plan and but in reality when this topology runs in physical Machin it's a completely different ball game reason why is like uh you might not be able to process the data in a single process or even a single machine and because of that like you end up uh having multiple instance of each one of those elements in the dag so the way the same topology will work in the physical machines is like you have tasks so you have the spout task which are several of them and similarly there's a part Tre bold task which could be several of them and uh you can configure each one of those stages or each one of these uh nodes in the dag by using a parallelism parameter so and you have a word count bold tasks as well so now all these tasks could be distributed across all the machines and they will continue to run and the size of your number of task is proportional to the amount of uh uh data that you wanted to process at every stage now one of the main issue is uh when a part tweet both emits the data on one of the tasks emits the data which task in the downstream it has to go to so that is where the notion of uh stream groupings come into play and uh so first is Shuffle grouping you have a random distribution of tles and it can go to uh any task Downstream and the fields grouping essentially takes the fields of some of the tle and groups them and based on the value of the fields it picks up the task Downstream then the third aspect is called all all grouping which is replicates the tles across all the tasks and finally like uh you have this Global grouping which sends the entire stream to one single task so most of the time probably like 90 95 to 99% of the time you use a combination of Shuffle grouping Fields grouping or one of those so now getting back to the word count topology how do you annotate uh the topology with the grouping information so as you can see from a tweet to a part tweet bold the Tweet can be processed by any part tweet bold which means a shuffle grouping will suffice there on the other hand when you go from a p tweet Bowl to a word count bow you should be doing field grouping because you want the same word from across different tweets to be end up in the same uh task so this with this introduction I'm going to move into why we did heren when we had storm in the first place so there was a lot of uh things that we uh looked up and especially after operating storm on several thousands of clusters um so what we did is like one of the major issues was performance predictability so it was storm was having issues in terms of how to predict the performance from a end to end point of you as well as the individual bolt and tasks and all the various stuff right so how do you have a system where you can predict your performance second one is like uh improve developer productivity so we found that storm jobs are very challenging to Def debug and that caused uh a lot of developer productivity in terms of trying to figure out uh a needle in the ha stack where ay what bolt and what instance of the bolt and on what machine it is running is causing issues how do you improve those so how to improve the developer productivity such a way that you can quickly get your uh box fixed and move on and finally like uh these are manageability I mean one of the major things that we had with respect to storm was we were running on different cluster on its own and uh so whereas we wanted to as a team we don't want to maintain our own clusters and everything instead we wanted to kind of run it alongside the critical services that serves Twitter and all the other services and it should be running as a part of those big clusters that we have so that we don't have to manage clusters and all the various stuff so that ease of manageability and the developer productivity and performance produc ability are the three main reasons why we started doing Heron so now what are the Heron design decisions first of all it should be fully APA compatible with storm and uh the reason why we are to be compatible with tomor there there are a lot of Legacy code that has been already on top of it and they are all in production and we have to make sure that uh we can migrate them seamlessly without having to do anything at all so without any code changes and any developers uh time so which means the notion of a dag applies topologies spouts and uh bols appli apply and all the apas are compartible the second aspect was task isolations so we wanted to uh have the notion where each of the spout and Bol are running on a task of its own or a process of its own so that you have the ability to do profiling and doing resource isolations and all the various other aspects that comes along with it so then finally we used mainstream languages which is like a C++ Java and python so because the team was uh uh the strong was return in closure we did not have enough tools for that language in order to uh make that development as fast as as we could so hence we resorted to writing uh some of the critical components in these mainstream languages so any questions before I go further deeper into heren okay cool hope I'm very clear so anyway so the first thing uh designation is like uh we got away from a schuer so as everybody know storm had its own set of scheduler for scheduling tasks uh in cluster of machines so we moved away from it the reason we moved away from it is because the Schuler have their own set of communities like Apache misos yon and then there are some Frameworks that are developed on top of these schedulers Apache Aurora then Marathon then we have e yes which is Amazon uh Docker container service there are so many of them so instead of uh introducing get another one and adding into the mix instead what we did is let's take a piggy bag on an existing scheduler and uh so whenever you submit a topology it looks like another job for the scheduler so that it can schedule the topology also alongside with other jobs so that's what we did so any topology that you submit this goes to the regular schedulers and that in turn gets the resources that each topology wants it and each topolog is run as a separate job so if you look at how the entire topology runs um so we have this notion of uh a topology Master which is responsible for the life cycle of the entire topology and that is running in a special container called container zero and when I say container it's nothing but a cgroup container that Lin supports so which allows you to have a nicely resource isolations and uh so that uh whatever the resources that you allocate for that container like X CPUs and the Y amount of memory that you like to allocate for the container does not get exceeded so that when multiple containers are sharing the same uh node you don't exceed your quota so that it does not affect any performance of the other topologies or other services that are running alongside in a multi-turn cluster so the topology Master runs in the container zero and the topology Master comes up first and looks at its dag and finds out how much resources that it needs to allocate and request that much of resources from the Schuler and uh the topology of also saves its state which is essentially the dag and uh some of the execution state which includes who launched the topology what time it launched what version of the Heron it is running on all kind of various stuff which is called execution state is all returned into a zookeeper cluster the reason being it's returned into zuker cluster is the container zero can fail the topology Master can fail in a distributed system so which means like uh if the container zero fails and the scheduler moves the container to somewhere else that container will come up and the topology will Master will come up again and will pick up the state from The Zookeeper so that it doesn't lose any state and the topology Master is not even involved during the data path it just is involved only in the control path as long as any changes in the physical plan logical plan is stored that is the state it maintains so then once the resources are acquired from the scheduler then uh these containers are scheduled by the um the data containers are scheduled by the uh schuer itself and each of these containers contains something called a stream manager and there is a notion of a metrics manager and there are i1 through I4 these are all nothing but instances and these instances are the ones that run your spout task and the Bold task remember I was mentioning about the word count uh topology which contain the Fe spout and the part Street Bol those task run in these instances i1 through I4 and uh the manager is responsible for moving the data around across the containers so essentially when data is emitted from these i1 and I2 and I3 so data gets into the stream manager from the stream manager it goes to the other stream manager before the data is delivered into the other instances so even uh if the instances in the same container want to exchange data they can bounce off from the stream manager as well so you can think about stream manager as a what you call as a data router so where it routes the data to the appropriate task from its source so that's what it does and the metrics manager essentially is one that collects all the metrics coming out of these various instances and the number of metrics that we collect is so huge I mean because whenever some topology goes into some kind of a state we need to know what happened with the topology right away so because of that we collect several hundreds metrics about the topology including how much tles it does process how much btes it does process what is the latency it experiences how much bytes it pushed into a stream manager all kind of Statistics whatever you can think of and um and how many times the container got killed and rescheduled how many times the process got restarted all kind of Statistics like that which collected and those things are sent to an offline system so question huh ask question sure number one coner is a Docker container so it could be mapped to a Docker container but in production we don't use Docker containers instead we use the we map a container to a Linux C group container yes in terms of Matrix uh it will be interesting looking from Whole task andri kind of collecting those but you have a way to cross cor across containers or so like currently we aggregate that information coming out of those containers so so but we have information about every uh spout and bolt and all the various informations also is available so so that you can look at whether a particular spout is the one that is causing the issue or a particular bolt is causing the issue you can do all of them yeah but you can that to application not no this is for only for operations yes so there are some user defined metrics for example as a user of a topology where I'm interested in collecting some metrics you can do that as well which is called user defined metrics we also allow that and that metrics also is pumped to wherever you tell it has to go to so that also we do that but for operational purposes we collect hundreds of metric so that if some topology that is in production they come running to us hey there's something wrong with it we should be able to locate the problem very quickly enough then so that he can turn around quickly right so yeah so thinking along the line of zip kind of yeah Zipkin is for tracing but this is more online immediately you can check the matrix itself yes so then um when the topology actually gets executed in uh in the physical nodes so this is what you will see so you have these four containers and uh some of the instance of the bolts and spouts are running in these containers and you have this notion of stream managers in this container right now what happens is once these containers come up the stream managers uh kind of uh what do you call interact or rather they do have uh what fully connected graph where each of the stream manager connects to the other stream manager and that's the only way they can start exchanging data right away so see any data that is a put into the topology from the uh spout S1 on the top left container we get into steam manager before it goes to the other steam managers and the advantage of using stream manager has uh is several aspects for example it reduces the number of resources in terms of the sockets and ports from order of n Square to order of K Square where K is the number of containers as compared to K is where n is the number of instances of various spout and bolt who directly connect to the respective Downstream components right and another advantage of isolating the stream manager code is uh you can write any kind of processing semantics that you would like to so like for example at least once at most once exactly once and uh you can even make one stream manager for each one of them so so so some of the her topologies that we running include uh the shapes of this form so you have simple topologies to all the way to complex topologies as I mentioned and some of the topologies are generated by a summing board which is a scolar interface that we run on top of heren heren you can think about as an engine and uh suing Bo is the API that we use and uh that API in generates these uh several topologies complex topologies some of the uh topologies that are smaller or these ones are directly return on the heren AP itself so yes so the mar must be instantiate ahead of time before you can do an so the topology the dag the dag shape should be instantiy the dag like for example resource also assigned so resources can be assigned dynamically okay so like when the topology Master comes up you can tell I need this many containers right yes but uh the dag itself has to be inchant because that is the contains of processing logic right so so her at Twitter has been in production for more than 2 years now and uh obviously I can't tell the exact numbers but uh we are processing tons of data and uh probably like the number of nodes that are running in heren is like probably few thousand nodes so it's all fully running um Heron itself so some of the use cases at a high level is uh real time ETL real time bi and product safety like fraud detection spam detections and all the various stuff and uh Computing realtime Trends and uh a real time ml in terms of model building and model enhancements and we also do some kind of realtime media as well in terms of processing and realtime Ops some of the operational data is continuously processed in order to spit out some um analytics and based on which quickly decisions are made so these are some of the high level use cases and um so I'm going to go delve into more of a couple of problems that we kind of solved in production and uh I just want to highlight those things with a short introduction to herin so the first thing is comprizing heren so so I mean the big data ecosystem is continuously evolving lot of new technologies coming in and going away and all this stuff right so as you can see the hetrogeneous system with respect to heren at least is there are multiple several schedulers like man schedulers unmanaged schedulers then the state managers which could be a zookeeper or even a local file system or even a Hadoop which provides some kind of synchronization mechanism and the job uploaders once you have your jar how do you upload that jar right that could be anything from an S3 to um a local file system or even hadu the in this case so this H ecosystem is continuously evolving in some fashion or other so how do you design a system such that your dependencies on this ecosystem is minimal and uh the core does not change and of at all so so when we did that like uh we looked back I mean one of the things that I always believe in is the computer science revisits itself in every 20 to 30 years years in some fashion so I said okay let's go back and look in what people have done in early stages right so one of the thing that we came up with was like um uh micro kernels so there was a notion of Monolithic kernels where everything was uh written as own single software that's what used to be the state of art at the time so where you have the hardware device driver schedular virtual memory then IPC file systems and even a virtual file system on top of it on where the actual file systems are Ma and they saw that why you have to do a monolithic instead they took only the necessary pieces and moved everything to the more at the application Level so so that's what called the micronel and micronel has the hardware and the basic IPC and a few things and the rest of the other things were moved to the more application Level and we kind of what is inspired by that and heren design also follows that so we have the basic in streaming any of the streaming the basic thing that you require is the ipcs essentially a way to communicate between processes in some fashion uh running on top of Hardware then everything is a modules like everything is a process and modules that you written on top of it so what happens is like you can go and replace anything you want I mean like for example I want to write my own highspeed instances and everything all you need to do do is instance interact with stream manager through a protocol and as long as you can impl Implement your protocol then you're done you can write your own instances as well and similarly you can write your own metrix manager already we have uh return metrix manager that can uh uh go to subscribe sorry send data to scribe as well as graphite and all the various stuff and uh whereas um the stream manager also can be replaceable because of the Plug and Play kind of architecture and topology Master's logic where you wanted to do something cool you can go and add that as well so essentially like uh every component is replaceable in some fion or other so we don't call heren as a streaming engine anymore instead in said we call this a more of a a streaming framework where you can add your functionalities that you like to so what are the advantages of these architectures plug and play as environment changes code does not change and also like multi language instances possible because today like some of the multi- language related API don't don't run in the native language uh instances instead um here you can write your own instance for example today the supported API is uses Java AP so which means we have a Java instances and if you want to write a python API for writing your topologies quickly enough then you can write a multi language one of the instances like python instances and boom then the python code will execute in the native python interpreter as compared to you having uh data coming from a Java process into a python process so that that it can execute all those other stuff you don't need to do actually in fact like um uh there is some work going on already to uh write them python instance as well so then multiple processing semantics that you can support then similarly since it's all componentized and multiple team members can work on individual component separately and it's very fast for development as well so the her environment so as I said like um U we have multiple environments for her in within Twitter itself so for example our development environment uses a notion of a local Schuler where you can bring up entire her and running topologies into the single laptop and uh it also uses a local state manager which is essentially like a synchronization mechanism and for that we use the local file system and the local uploader where again we use local file system to upload your jobs and run it and that is when our primary what do you call development environment where we I trade very quickly fast enough and it runs as though it is running on a distributed system itself so it's easy to for example if I introduced a bug in Steam manager then I can just GDB into the steam manager boom and I can debug that right away so it's very fast and uh easy for development second one is like for cluster and testing we run it on m/ Arora schedular and uh with zuker State manager then we use HS for the uploader and that's is used for testing purposes in a distributed systems and our production runs in uh Aurora schedular which is the uh manage cular that is uh written on top of misos and we have a zookeeper State manager and the Packer uploaded Packer uploader essentially manages versions of topologies and all the various things for example your topology might go through various versions depending upon the functionality bug fixes that we added and it can manage the versions of the topology as well and that's what runs in production So within Twitter contacts itself we have multiple environment and there are other groups which have taken Heron and running it on AWS as well have not included them but they're running it on top of AWS as well too so now why Aurora I mean Aurora has been one of the very uh widely used uh scheduler within Twitter and it manages hundreds and thousands of missiones uh so a lot of critical Services run on top of Aurora itself so so the heren runs on in the context of a Aurora in a multi- cluster alongside with the other critical Services as well so then uh it allows you to provide authorization and expressive DSL to write your jobs and it's also very high availability and scalability because uh there's a team behind it which manages the whole show of for aora and lot of inbuilt metrics for monitoring and debugging as well so with that uh introd this uh short uh this thing discussion about uh component iing here and I'm going to go into the next problem that we face in U production it's called stragglers and back pressure um this is a very interesting topic actually um so stragglers are the norm in a distributed system when you have a distributed system there are several reasons why some process might be slow I mean especially some process that is running a part of as a single job could be slower there are three primary reasons for it one is is a bad host like uh 80% of the time we say he by the way like the machine that job got schedule is bad because it could be having some Ram errors or some disc errors or whatever might be the case especially when you're running on a a few hundred, node clusters some bad apples will pop up here and there so that's a commonly observed problem why uh stragglers start occurring the next one is execution skew which is essentially like a manifestation of the data hitting one process a lot of data hitting one process because of the hotkey problems then uh another the third one is what we have seen is inadequate provisioning so if a job requires some uh uh x amount of provisioning in terms of resources but you have not given the enough X stragglers will start occurring as well so now what are the pro approaches to handle stragglers so the first thing is uh send us to whenever your straggler occurs or whenever there's a straggler in your job you can uh send uh data to that straggler but it'll slow down everything so now a simple strategy is when the sender figures out I can't send as faster to the receiver which is the straggler I simply drop data so that one approach second approaches is like slow down the centers to the speed of the straggler itself so that everybody is going at the pace of the struggler so in this case uh like um there will be predictability and accuracy and all the various stuff and detect stagl and reschedule them so like proactively detect stagers and reschedule them in some form or the other so the drop data strategy so what happens with the drop data so it's very unpredictable so uh it if the machine is going really slow you can continuously drop dat data and there are instances that we have seen is a 2 to 3x inaccuracies so for example stor was uh doing this drop data St strategy uh with that what happened was like when the data was being shared with some of the potential Partners they saw 2x to 3x inaccuracy During certain times and that become uh affects accuracy quite about especially if you're trying to share some essenti critical data and also it gives poor visibility in terms of where it is being dropping and all this thing because if especially if a job is running on a few hundred machines how do you know which one the guy is dropping data which is very difficult to locate it so the Slowdown Center strategy so in this case it provides predictability and uh depending upon how the stragglers or the stragglers are not present it process the data at the maximum rate and also reduces recovery times whenever some uh thing fails immediately it will pick up uh it it gets rescheduled on a different node and it is going really as fast as it can so then it can nicely handle temporary spikes so if you delve more into uh the stategy uh let's take a simple linear topology which has a single spout followed then in turn goes into bolt B2 and uh that in turn feeds into a bolt B3 and another one bolt called bolt B4 now so as I said like uh there are four containers that are running these pouts and bolts and you have these stream managers uh which are running in those containers and they are the ones which are providing the data path for exchanging data now let's say for the sake of discussion the B2 is going slow and the moment when B2 goes slow the the steam manager detects that the B2 is going slow then uh it will send that uh what do you call initiate back pressure or initiate uh like a slow down strategy to all the other stream managers so now the stream managers when uh the look at their instances on this respective containers and when they see a spout they say that we are not going to accept any more data so what we are trying to do in this strategy is like uh the source of the data is the spouts so if you're not taking any more data from the spouts then the uh pace of the topology could be controlled right so so by choosing when to pick up the data and not to pick up the data you're going at the pace of whatever the straggler is so which means uh everybody slows down so everybody slows down to the pace of B2 so now when B2 picks up speed then everybody will start picking up speed because uh there will be another message that sends out saying that hey by the way back pressure is relieved everybody can go at that at a high rate right and uh the way we do that is we have a what notion of a uh Watermark High Watermark and low watermark in buffers and by using those Watermark buffers you can I just when to kick off the back pressure and when to relieve the back pressure so the back pressure in practice when you looked at it like um so if back pressure is kicking in uh so heavy then you will fall back in the read part of it because uh you're pushing the back pressure to the source now once you hit the source and if the source is not consuming enough data then you know clearly that you're falling behind and one of the measurements that we make is the lag so how much is the read versus the right lag is there and that is proportional to the whether the topology is going at the input rate or uh is lagging behind so that is a measure that we use in order to figure out if there is any issues in the topology or not so if that is the case uh sometimes what happens is the back pressure might not recover so we manually shoot down that container and the container gets rescheduled automatically restarts so so back pressure in practice so most of the scenarios back pressure recovers and without any manual intervention and which is a nice thing and uh there are some instances where we Face sustained back pressure and that partially attributed to bad or faulty host and uh irrecoverable GC Cycles so one of the thing that we are looking at into it right now is uh doing anomalies detections Within These uh containers and the spout instances and all the various instances so that we can detect this bad and faulty holes then we automatically shoot down that container so that the container gets rescheduled somewhere else without any manual interventions so that is the problem that we are looking at in detail and sometimes user prefer dropping off the so some people said no no I want to be at the top of the queue always if uh the processing is not done on time we might lose some other but I don't mind being inaccurate so so for that we use sampling based approaches so since um uh and how do you detect bad and faulty codes we use another Heron topology to detect them so the hundreds and thousands of machines that are there and uh we have a heron topology that detects those data coming from those and analyzes data see whether that topologies so not topologies the any bad host is present then you go and uh put the feedback and take it out that machine out of service so I'm going to skip the load uh shedding section and instead I'm going to go into uh what do you call uh the resource usage of her so one of the things people have talked about is yes can you keep the question a little bit later I have only five minutes okay so yeah so the her and resource usage uh we took a two node topology two one is the even spout and another is an aggregate Bold and the even spout takes 60 to 100 million tles and filters the tles to 8 to 12 million and uh then we do a flat map which essentially blow up of the uh tles essentially like once we filter these tles are breaken up into five times so that um the amount of data that is coming out is 4 to 60 million Tes per minute and that is given to the aggregate bold which Aggregates every 1 second and it outputs around 25 to 40 million tles per minute writes it to redis now let us see how the resource usage was so heren requested 120 resources to the schedular but the actual usage was 30 to 50 CES and memory requesters was 200 GB but memory used was just 180 and the red is we had 24 CES requested but hardly used 2 2 to four CES and uh now zooming in where the St the resources are actually going so if you look at it 84% goes into the spout instances that's very unintuitive where the data that we fetching from Kafka whatever it is that those are the ones which are taking the huge amount of time or the huge amount of resources the BS hardly do 9% and her and over is it's hardly 7% so now profiling into spouts to figure out where the actual time is going so we can see like the D serialized cost is 63% and uh paring and filtering take 6% and taking it out of the Kafka the Kafka iterator as well as the Kafka fetch take 16 plus 7 that is like around 26% or some like 2 4 + 23% so so it says the fetching data which is 63 + 16 + 7 that's close to 80 to 85% of the whole spout is gone to fetching data itself If you eliminate that whole head and if you take the binary raw binary then we might even shave off all those things so now profile BS again like going to writing data into redus that takes 68% of the data remember the overall grand scheme of things uh bolt is taking much more lower compared to spouts itself then again serialization and dilation takes around 20% and some of the aggregation data transport takes only small amount of percentage so effectively like fetching data cost you more like 61% and ual logic takes 21% Heron usage is 11% and we have room in Heron to make it even a 3X and Forex faster but uh this is adequate enough to most of the processing at this point so now one more numbers like in terms of uh one of the um topology that mimics production topology it counts the realtime active count where people are interacting with Twitter so some of the CP usage as you can see number of codes used versus heren versus uh when you are at least once and at most once the number of codes is significantly reduced in her and itself so the overall save like 3x to 4X worth of cost in terms of the number of machines and CES that we use and uh if you want to learn more so we have a bunch of papers so stor at Twitter which was in Sigmar 2014 uh uh Sigmar 2015 Twitter herent then this is a new paper that is coming out then if you wanted to do more a compendium of realtime analytics so we did a tutorial in vldb 2015 this is 180 slides of lot of hardcore uh in-depth content so if you are interested please go and take it's available in slides share so you can take a look at it and that's all I had um I think I little bit rushed but uh hopefully everybody got the gist of it yes please thank you