scala.bythebay.io: Ashvin Agrawal, Avrillia Floratou, Karthik Ramasamy, Twitter Heron on YARN/REEF
you yeah hi everyone so I'm Aurelia and today along with husband and Karthik we be talking about a very exciting collaboration between Microsoft and Twitter on large-scale streaming systems so we'll focus on Twitter here which is a distributed streaming system that has been developed in Twitter and has been deployed in production for the last two years Microsoft is actively participating in this project so will also focus on the contributions of Microsoft in this in this system so the outline of the talk is the following I will first start with an overview of why streaming is important and why Microsoft is interested in streaming I will follow sorry something is anyway I will follow with a discussion on the modeling but the architecture of here and the most important points and what are the design calls of hearings architecture then as Ben and I will talk about two important modules of the system namely the resource manager in the scheduler Microsoft is active actively contributing in this two modules then Karthik I will present an experimental evaluation of here on and how it compares with other open-source dreaming ends it's like storm to highlight the efficiency of the system and finally we will conclude the talk so why we are working on here and so as I said before hearing the system that has been developed in Twitter and has been in production in Twitter for the last two years it handles pretty much pretty much most of the Twitter's workloads so twitter has very demanding workloads it requires to find real time friends in the tweets or real-time conversations or recommendations for example when a user types a tweet or discusses with other users then in Twitter they figure out three they need to figure out really fast what recommendations what ads they should show to the user and also they do real-time search so here on basically is the basis of Twitter's pipeline for all these applications it can handle Billy so the events in real time so why is Microsoft interested so Microsoft has a lot of streaming data we do real time analytics real time machine learning we have etl applications where basically we read data from one system through a streaming ends in we perform some cleaning and some filtering and we store it in another system this has all to happen has must happen at real time then we also have monitoring and alerting systems based on streaming infrastructure so for example in our cloud we collect matrix about our VMs our machines if there are failures if they're an expected events and then in real time we need to alert the user of what is happening all that is done through streaming pipelines and then we also have iot applications where basically we collect data from sensor networks we perform some processing and some analysis this is again going through streaming pipelines so Microsoft has very streaming engines but when we look into open source we see here own as one of the most promising ones for various features that we will discuss later in the talk so for this reason Microsoft is actively participating in the open source project and contribute several modules so now I'll give a brief overview a high-level overview of hearings architecture so hearing has been designed with a goal of extensible of being an extensible streaming ending what that means so when he owned was open sourced we wanted to have higher adoption so that means that we need to integrate with virus big data platforms so for example in Twitter here on operates on top of the missile scheduler however other companies use the yarn scheduler this is because they deploy Hadoop per hive on top of faith defense data using the yarn scheduler we wanted to be able to incorporate here on in this environment that means that we want to architect here on in a way that it will be able to adjust in these different big data infrastructures without providing without making making it a better for the administrator or the developer secondly we want to adapt to different application requirements so for example you have users that want to write application needs of our scholar or C++ or Python we want you to be able to handle all these different types of applications third we want is of the ease of deployment so we want people to basically download here on from one code repository and install it easily in any type of infrastructure they have we don't want to create separate depository specialized for different data big data stacks so we don't have a different repository for yarn or different depository for mrs. it's all in one code base and finally we want accessibility we want the users to be able to extend the modules of hearin that they want so that they can satisfy their own application requirements for example a user may want to implement a new resource management when the source manager that optimizes for network communication in history me environment we should provide the api's to allow that so hero has been designed with this goal essentially its component ain't hearin is extensible and all the components communicate through well specified protocols so now I'll talk a little bit I'll give some terminology about here on so it's open here on is essentially described using a topology at apologies directed acyclic graph the vertices are essentially the computation and the edges represent streams of data tables moving between different vertices so the vertices are of two types spouts and bolts so spouts are sources of data data tablets are for the topology for example we have a Kafka spout or a postgres scout spout or am I sequel spout and sensory reading data from different systems and the balls are essentially that represent the computation that the user wants on the particular datatype priest so balls can perform filtering aggregation join or any other user-defined function and users can write bolton spouts in different languages they can write a Java and C++ in Python so we're pretty flexible on that so this is an example here on topology as you can see here we have two different types of spouts and five different types of bald and they connected to each other with different communication patterns so essentially when the user rights are here in topology has to specify 2nd specify the functionality of its pouting bolt and how these are connected and we can support any type of directed directly the cyclic graph so a simple but popular topology is that word count apology we're basically we have a spout that reads like a live stream of tweets and then we have a bolt that takes these tweets from the spout and perform some parsing some filtering and essentially figures out which words are present in a particular tweet and then sends this word to and this stream of words to another bolt that basically aggregates the words and finds how many times its worth has been present in the tweet so for example the word soccer has been present four thousand times so far weld cup 1 million times and so on so now I'll move on with the hearing design goals so we have several goals when designing here on my need to say at this point that before here on was deployed in Twitter storm was the system that has been used before for handle Twitter's workload but storm has ever had several problems and the that's why it was replaced by here the main reason is that hearing improved a lot of performance and also provided isolation which make it easier to debug and maintain the whole system however one of the goals I was to be fully API compatible with storm so basically here on can accept an existing storm topology and run it as it is so we kept the same API second here on provides task isolation so it's bald and it's part of a hearin topology run on its own so if there is a failure or some problem then that does not affect other components and here on is able to recover easily i will talk about that in a few slides then hearin as i said support many languages c++ java python scala then another important feature of hearin is support for back pressure which means the following if a here on instance cannot handy for some reason the workload then it notifies the center basically the other instance the sense topless to eat about this event and the center slows down so that is very important because we avoid failures by having topology self-adjusting here on also supports boxing of tuples so which is which and the bad size is also a parameter configured by the user which means that if a user wants to improve its his latency they can remove he can create small boxes if he wants to optimize for throughput you can create bigger boxes so this is pretty configurable to handle different types of applications and finally here another goal was efficiency so we want to have efficient resource consumption and also very good performance so to satisfy all these goals we created here in using an extensible architecture as you see here the main component of the system is basic inter and intra process communication that's essentially the general of the system every other module of the system is an extensible component so here in blue you can see the different modules so we have the scheduler the topology master the stream manager here on instance matrix manager resource manager in state manager so all these modules are communicate with each other with well specified protocols but they're pretty explain extensible so for example as husband will talk about the scheduler can operate on top of a Mesa scheduling framework or on top of a yarn scheduling framework or the resource manager can implement different resource policies for for a topology so these are all very extensively and we are also working on making more extensible other components of the system so ano worried about how hidden works so when I user-submitted apology there is a scheduler for basically coordinates how many resources should be assigned to that apology and it's responsible for bringing for vetting these resources from an underlying scheduling framework like mrs. yarn and so on and then basically determines in which machines this this topology should be execute and schedule these typologies so basically that that's the very high level overview of the system if we look in more detail into a particular topology I will see that its topology consists of a topology master a single topology master which essentially tracks the topology through its lifetime the time by the time it gets submitted until the time it gets killed the topology master track the state of the topology and there's there's a state manager which in this case is a zookeeper cluster it could be something else as well this is also an extensively module but basically keeps the state of the plan of the topology meaning with instances are up in wit machines they're running or this is store in the zookeeper cluster and what is the execution state of the topology is it running is it active is it deactivated and so on and then the core processing is done in containers its topology consists of a number of containers and within each container we have a set of here on instances here you can see them I 1 I 2 I 3 I four but essentially represent user code written four bolts and spouts so here we have four instances and each of them has a parallelism of two then its container contains three manager with a crucial component it is responsible four hundred communication across the instances within one container but also across containers and it's essentially the module that routes the tuples across the topology and then its container also has a matrix manager that collects matrix about the instances in a topology for example latency matrix throughput matrix fave number of failures and so on and reports this matrix to the topology master and then it could also report to other systems like graphite and other monitoring tools so here on has fault tolerance so what happens when an instance fails when spells in a container then there is a process called the executor that basically restarts that instance what happens when a container fails when a container face then the scheduler yarn or missus is responsible for starting this container and restarting the processes within this container and when that happens the stream manager of the container contacts the topology master says I'm here and then the topology master creates a new plan for the WG taking into account the new stream manager it populates that plan in the zookeeper cluster where every container pics that plan Anna just basically takes it without the new container to do the other routing of topics and when the topology master dies then again the scheduling framework can restart this container in this process so that's the high-level overview for tolerance so as I said before the topology master is responsible for monitoring the state of the containers it responsible for getting matrix and routing them to different components of the system and it also signs roles which means essentially that there could be different groups in a company one group focusing on security and other group focusing another application they have different roles and we can distinguish at apologies and matrix based on a particular role maybe karthik knows more about that will be secure people and I'll do so the three manager is the components that performs the routing of all the topless in diplomats back pressure which is very important in hearing because we avoid many failures because of back pressure and it also performs a knowledge management in case the user have a topology have an ambit Oxford apology these acts are going through the stream manager this is an example so here we have four containers for a topology I join managing a specific set of instances and its container there's a stream manager you see that the instances do not communicate directly with each other essentially the streamers and in each container contacts three managers on the other container and handles all the communication so this is really important because we can isolate the communication problem from problems from the instance problems so it's very easy to debug and figure out what are the what is the potential problem in the system and then the hidden instance as I said before it's essentially the user code in trance one task only one task either a spout or a boat so it's one run on its own JVM there's full isolation there with which facilitates better maintenance of the system then the instances expose expose both a storming here on API so you can take your existing storm topology an unknown hero and the instances collect several matrix about their lifetime and then they expose them to the matrix manager which and then go through other tools like the topology master or graphite or other monitoring tools so now I'll talk a little bit about one important here in module that Microsoft is actually actively contributing to and then as we will talk about another one so I'll start with a resource manager the resource manager is a very important component it is essentially when a user submits a topology and says I have this number of spells these number of bolts and this communication partner the resource manager is responsible for assigning resources for the spouts and the balls and figuring out with spouts and which volts should go to its containers and how many containers so when this parting plant is called packing plant when this is generated it is provided to the scheduler and the scheduler that is these resources from a scheduling framework like yarn on missiles so here you can see general topology with four containers we have one spout in three bolts and as you see the parallelism of the spout into the bolts is for and the barrel is for one of the bolt is three so the user is able to specify how many instances of its belt and bald see ones and then the resource manager basically is responsible for figuring out how to place the instances to the containers so that every instance has the resources it needs the amount of memory disk and cpu and also optimizes based on specific metric in this case we use the round-robin algorithm which basically optimizes for load balancing however this module is very extensible we could have used another algorithm we have one based on bin packing which basically tries to assign the instances in a way that minimizes the number of containers used there could be other algorithms for example based on graph partitioning to minimize the network communication among the instances and so on so the the most important thing is this component is very extensible and the users are able through it we implementing the very basic api's are able to optimize their particular application so another important aspect of the resource manager is that it handles resource scaling so when a topology is running the user here on basically supports scaling of the topologies either scaling up or scaling down so the user can actually say I want to remove an instance of ball tool I want and I want to add an instance of spot one and an instance of bolt three so we don't have any disruption in the topology as this happens the resource management manager is responsible for allocating the new instances two containers either existing ones or by creating new containers and no this happens without disrupting at all the topology as it runs so this is a component that we for which we introduced well specified API so essentially the user or the developer has to implement four methods and if wants to extend and not use an existing algorithm for resource management it's a full extensible module and it works with both homogeneous and endogenous containers what that means is that some schedulers like messes can only give to the user set of homogeneous containers with the same amount of resources same amount of ram cpu and disk whereas other schedule is like beyond can provide a terror genus resources so containers with different types of of resources for a given topology so the resource manager is able to distinguish this these two types of schedules and optimize accordingly so now I will pass to a husband and he will talk about the scheduler oh sure so for it yeah so the question is if in the resource manager we can run 1 a 1 per node or multiple so the resource miles in May is a centralized component / topology so there's one resource manager / topology so these yes it's you can run the good thing about that is within one cluster you can optimize different apologies for different things so one you can optimize for load balancing another for I don't know resource allocation and so on so you can have different algorithm in topologies yeah I alcan I'll admit that I didn't expect so many people on sunday for a yarn talk thanks for coming welcome if you have any questions just raise aram i'll try to answer it if i don't know the answer will go for a beer and we'll resolve it there ok so it's very good question that i asked every day and clapping to show up but yeah I'm Ashwin I work for sisal in Microsoft and this is the labs we work on distributed systems and when you talk of Microsoft to talk of scale and when you talk of Usher and building and all that you want the speed the result should come as fast as possible right if there is a person who is logged on in your private cluster and you get a notification after three days after a batch processing system finds it then it's too late by the time lot of things have been lost if it is billing if it adds camping you're running and the results you get after a week that your campaign was not as successful or you did not focus in a particular area where you could have focused then it's too late so you want a real-time system and when talk of Microsoft I'm talking about thousands and hundreds of nodes which are which will be solving these problems right I will be talking about he on a on is a common the shared compute platform which is used extensively in Microsoft there are two models in which it is used either it could be is used as a private cluster for a business use be used by a partner or something or it could be a service offered on top of yarn for example it could be hidden which is offered as a service in a and which is managed on e and by Microsoft but the common substrate AC on how many people from with yarn awesome if you have good questions you know right so typically this is how it will work in a yawn cluster I will have a resource manager this is the central this is the first entry point we on plus tight manages what kind of how many resources you have what jobs are running how much Headroom is available in the cluster and then on each of the worker nodes there is a node manager so I have resource manager node manager the node manager is responsible for running the processes on the container so when I want to run a task when a job I will ask the source manager to allocate resources the resource manager will contact the node manager asks for headroom and then allocate processes in that so this is a typical flow of the typical flow control flow in a yarn cluster I'll skip the details on how the higher ability is guaranteed by yarn for a resource manager there is a this is a higher bility plan if a node manager fails there is some plant by which is offered by yarn if you're curious we'll talk about it but i'll skip that into into interest of time right now i'll skip this slide so as I said like in in Microsoft we have a shared computer platform which is he on base right and assume that they know you sent this yarn manager in there a couple of node managers and I have some applications which are already running on my cluster so why this was important is that it is no more efficient to have a private cluster per service in your in your data center it is not efficient it is expensive it is very hard to manage it is vulnerable so I'm talking about shared resources all the time and hence what I'm saying is that I have a yarn cluster and trying to run which is running some jobs and i will add head on on top of that but I'm talking about 10,000 nodes the installation has to be simple right do you want to go to each of these nodes and install binaries now so when you talk of the current deployment it's very simple the installation is actually preparing two nodes so there is a tracker using which the user can see which topologies which workflows the user is running are available right now what resources they are using and there's a client which is the actual entry point from where you authorized yourself for security reasons and then for quota management at how much resources are available to you as a user to use on the yarn cluster and that installation happens on one node so the only two nodes on which the the admin will deploy herons head on binaries and after that the head on integration and yarn will take over and it will be pretty simple from there so we have talked about different components of a topology on heron and what happens when a user says I want to submit a topology I want to submit a business workflow so at that time the Heron client will request the resource manager for a new job the resource manager will find resources in the cluster wherever available and then it will start the job there so what happens in a yarn no system is that there is for the cluster there is a social energy which is the admin of the cluster and for each application running inside that cluster there is a there is a am application master which is responsible for managing the applications life cycle so i can have tens of thousands of applications running on yon and that is how on scales to tens of thousands of nodes because if the resource manager yard resource manager was managing each of these applications like in Hadoop like in MapReduce one it wouldn't scale right so the when the application is the topology submitted it starts an application manager the application manager in addition to yarn work it will also manage the topology master which is the herans parent apologies master and also run a scheduler which is one of the resource components which I really talked about earlier subsequently this am will talk to the resource manager and get the containers I located in an event that this topology okay there to one important aspect here the container is requested by the topology master could we are the same size or they could be a different size what I've learned that if you use or owner cluster and all that the containers have to be on the same size which means if the largest container is 10 gig and I have 10 other containers which are one gig only but I will end up you over utilizing my cluster in that case but in yarn you can have heterogeneous allocation of this sources right and then more more and more topology scadman submitted what did we do for yon integration so there is a couple of options I could have used the native jaan AAP is wherein I have to rewrite the M communicator the node manager communicator declines and all that I have to handle failures lot of other options which is the first implementation of yarn which if you have written an application you know it's too complex too hard and its most vulnerable so what they're two solutions available and open so these are abstractions on top of your er API one is called slider and the other is Apache reef both are open source and they are to simplify the way you manage the applications on yarn we explode all the three options and then we decided to use apache reef so why did we use apache reef one example is that suppose I'm writing a very simple application on yarn i will happen right one thousand lines of code to get the application warning working if i'm using apache reef then i will only have to write hundred lines of code in Java to get it running and why is this so because the reef infrastructure which is like anything in the blue color here that takes care of the protocol implementation implementing the code which is which you don't care about all the business logic is what the user has to implement and that will take care of input the applications business flow and what happens when you talk to the resource manager that is all taken care by the reef and infrastructure there the the point I wanted to emphasize here is that when you talk of shared compute platform the talk of the advantages and all that but existing applications don't work as this on a shared computer platform there something has to be done and it is as of now I think it's not easy and that is where you talk of all these additional frameworks for interaction interacting with the compute compute platform ok so just for talking to yarn do I have to learn another platform another framework looks like Network right but now there are other benefits of using apache reef what we found is that dependency injection which is very very common in Java world these days it is that problem has been solved is it is integrated into the framework so if you use the hilly framework you get dependency injection configuration management static checks and all that because when you deploy the topology at runtime we do not want to see the failures you want to detect them as well as possible in the testing phase in unit testing phase and that is what tank offers then there's way that is what we've also brings in wake which is for even based programming to reduce a memory footprint the Radio City Layton sees the number of threads you are creating group communication all this and for people who are into machine learning you know the importance of running meant the importance of memory state which have you which you have created in a container right you're quite you get a container you get a host where you build some state and you want to run your subsequent iteration of the same node so that you can you can reuse the memory right and that is where the the first are in reef comes from it is a retain able part of it so when you run it when you acquire a resource from the resource manager you retain it you can reuse the memory you can reuse the state which you have built in there so three brings in lot of benefits if you're writing a young application I would definitely say that you give it a try I'm not a storm expert I wanted to do some experiments here on comparisons with reef and I tried couple of things I was not able to get storm working on yarn then it was sunday i thought let me give it a reef a try in half day i was getting i was ready to than all the experiments storm with hundreds of containers i was able to get it running on yarn on my shared clusters all security user authentication authorization they can be taken care of so awesome awesome framework tried in addition to hell and obviously so i mentioned the other two flavors in which we use this in microsoft one is as a service and the other other is HD insight HD inside is a hosted hadoop service which is available to external customers also right and we have made sure that all the building blocks which are required to run it on HD inside are available what it will do is like if you want to try Heron today you can go to go to assure you can create a HD inside cluster on which you can try Heron today it can scale to your demands you can test it you can see the logs you can see the flow of the performance whatever you want matrix everything is available there the code is available in the open source and if you can try it and give it let us know it's elastic as I said and in addition to that only heron will you need other components also to complete your business workflow you need a Kafka event queue you need something for for persisting the processed of data which would be HBase it could be other other stones so data lake event Kafka all these components are available in HD inside you plug in you can pick up pick your tiles you can play the solution and deployed in Beijing site very fast at saucer so this is we use this extensively for testing storm and heron both in for our test cases I like thank you so you're seen the overview of Haran and how Heron has been mapped into Apache reef for running in production workloads at Microsoft so internally within Twitter we use it in the missus framework miso / Aurora and it also runs on me show / marathon so when we announced a heron in 2015 so we did some performance numbers with storm at that point and we found out like we were like around 10 to 15 x faster than storm from a throughput point of view similarly we were 10 to 15 X improvement in latency as well then the stall folks picked it up and said that her we are going to improve the perp further the performance of storm and then at the time when Heron was announced in 2015 we didn't even do any optimizations at all in the sense like whatever the code we wrote and deployed into production that itself was 10 to 15 x faster so then we looked into heaven and see what we can do in not optimize it for that so what are the optimizations that we did was in a component called stream manager which you all saw in every container at Ron's right the stream manager like then we looked at all the profiling and everything then we've figured out we can move around a few copies and after that we were using protobuf heavily and the proto buff was doing lot of allocation and de-allocation of memory so whenever every message comes and goes so then we retained a memory pool and so that memory can be reused so with this bunch of optimization we got a huge speed of improvement in terms of throughput as well as latency so i will talk talk about a little bit on the performance so the performance settings is something like this where he of the components potent bowl and the number of containers so the number of containers were varied from 25 to 200 some of the experiments will highlight the numbers on 25 hundred and two hundred and some of the experiments will highlight on 25 50 and 75 so the spout and bout both parallelism or sessions lay like a each container will consist of one spout and one both okay so so now when we compared with the storm latest version versus the hare and speed up numbers as you can see this is like a huge improvement in throughput itself as you can see like this is in the at least one cymatics where you get another five to six x consistently over the storm new version and also the latency is much lower in herrin as well so one of the experiments we have found that Heron can do 13 milliseconds round trip between in a two-stage topology where the topple goes from the source to the boat then bolt back to spout especially in acknowledgment we can do in 13 milliseconds so there is a relationship between how there are a couple of tunable parameters that we have called the cache plane frequency words as a what something called the max pout spending so they both affect the true Putin latency and for every combination of this MSPs and the cache trains frequency there is a an optimal throughput and latency that you can get and the latency could be as low as 30 milliseconds and throughput is can be even higher okay so then so then we did the impact on the stream manager optimizations and as you can see we were able to without optimization that the first version of Heron this is the new version of Heron that we optimized and as you can see we are doing close to 11 billion couples per minute so that's a lot of tuples that you can move in less than 17 missions so again like well so there was a slight increase in the soup usage because of the fact that the stream manager got efficient and we were pushing a lot of data and because of that there was a little bit increase in the code usage then we computed what we call number of tuples per minute per code to see whether it's cost efficient as well as you can see on a per code we are still again 5x faster than the previous version of the Heron itself so so now if you do a head and usage breakdown where we wanted to see how Heron is very highly efficient compared to the rest of the other things that goes on you have as you can see sixty percent of the data is sixty percent of the core usage goes into fetching data from my external data sources like Kafka orchestral because of the fact that the data is serialized in some form so the cost of deserialising and the cost of fetching the data out of this messaging system is the most expensive component of the whole computation that you do inherit and the twenty one person of the ghost in the Lewis a logic where you do some filtering followed by some kind of aggregations and all the various computations that you do I don't even person of the time goes there a late person of the time go or the eighth person of the resource usage goes into writing the data to the next onal system like the red is or manhattan or a key value store anything that you want right so as you can see the hair and they use usage is just love and passion which is hardly anything compared to the rest of the other things so if you want to learn more like we have published two papers there is one more paper that has published two and the two more also one more has been submitted as well so once it's accepted we will kind of point out those those those papers but these serves a good good interaction for the storm and what are the problems in storm that occurred that drove the design for haren then finally like if you're any questions yeah you can be short on our mailing list so interested in Heron Heron is open source contributions are welcomed but there are a lot of big companies already started working on it and if you want to follow updates you can follow it doesn't screaming yes you want exactly once yeah so exactly once is on it huh okay so will you be interested in it using this for applications where I the depth checks could be deposited right so so so the in a streaming world exactly once getting it right is extremely hot or even probably like there will be always some amount of in accuracy you will be there always because of the way the nature of the beast itself because if you look at I have one thesis if somebody wants to prove it you can prove it but this is a hypothesis that we have observed over a period of time running the streaming volume how close to real time you want and accuracy so out of the three you can get to probably not three similar to cap theorem in some ways but it is intuitively sounds interesting but unless we can prove it right but i think is like so because of that even if you try to do exactly one system there are scenarios for example in the case of streaming joins and much more later I was then what you can handle so getting exactly once is extremely hard or it might not be even possible to do that so we are working on some kind of a state mapping maybe if you look at Tom Tom had this notion of a trade and how many of you know trident so it do use to do exactly once I need to add a lot of issues and production so that's why we had to kind of a decommission it now we are looking at different way of doing exactly once in the context of a state check pointing and retrieving and we will be announcing in a probably a couple of borders so that is the idea and if anybody interested in heaven riding ask all I API you're welcome to do it because somebody really interested in scholar we use kala inside but today but that AP is not exposed but if anybody is interested in doing it in open source please welcome and talk to me because so that's it you