scale.bythebay.io: Sanjeev Kulkarni & Sijie Guo, Streamlio: Towards a Unified End-to-end Solution
thank you morning everyone my name is Sanjeev we from stream leo and today we're going to present our vision towards a unified end to end real-time solution so just getting started right and basic stuff why real-time you know data is coming in data is most actionable and most valuable when the data actually arrives that doesn't mean that it doesn't have any value later on but the value of the data continuously keeps on decreasing the faster you act the file the nimbler you are and therefore in any kind of business decisions or operational decisions or any kind of the decisions that you want to take the faster you are the better and therefore you know real-time is very important all enterprises these days are moving towards some kind of a real-time you know even driven architectures kind of doubt as well with what we've been doing extremely oh how do you do this today well this is the state of the art you know you have some data streams coming in from the left and and this is the lamda architecture the the you know made famous by the building the company that this building belongs to Twitter of whom we also belong so you have a messaging queue so the data comes in data is actually split split into two streams so essentially there's a duplication going on one part the top one goes to the real time the speed layer or whatever it is called consists of a messaging queue of some sort you know it could be Kafka it could be rabbit and cube pulsar D log and then there is there is some kind of a stream processing compute that happens over it you know typically you know Twitter no storm kind of started it will move down to heaven that is flowing spark these days case streams and so and so forth right so a bunch of systems where you know you you can you can do things like you know you want to count how many tweets are you receiving or you want to do any kind of more difficult computations like trend analysis and so on so forth and all of those things are modeled you know code encode as topologies handled by the frameworks stream computing frameworks and these are some of the examples and and then after you do that you actually want to use these quote-unquote results like you know how many people visited my my tweets how many people retweeted my tweet or something like that you typically keep your result in some kind of a fast access you know like a cassandra key value store Manhattan in Twitter's case or Redis or you can actually store this in a database somewhere you know for later analysis and so on so forth so that's the that's the speed layer right and analog this at the same time what's happening here is that you know there is the data is actually copied over to long-term storage / the batch the batch world typically comprise these days of HDFS as the base edge layer MapReduce more recently SPARC and so on so forth right and and then the the reason why you sort of do that is because you know the speed layer is inherently lossy in some sense now it doesn't really capture accurate data The Bachelor is supposed to be accurate and therefore you want the latest results with some data accuracy loss at the same time you want the result acting accuracy to be bounded so you kind of combine it at the query layer towards the right to get the results from both the speed layer and the batch layer and and that's essentially a lambda architecture for you right and and for the minute you sort of see oh my god this is like a patchwork of systems like and a lot of things a lot of things can go wrong here you know first of all it's it's duplicated effort every data is stored at two places and acted at two places and you know and the computation is expressed in two different places and so on so forth right it's actually pretty heavyweight these days are the other you know flavor is the cop architecture where say you know what why are you doing everything you know everything should be done in just one and that's real time right so data is coming in you know did you put into a message queue and you have some kind of a streaming engine the same seeming engines as before and then and then there's a serving layer right what this kind of glosses over is that this is good for just real computation now what if you actually want to do analysis over yesterday's data or last month's data right that's kind of glossed over you know there are many ways of sort of doing that you can say you know what okay copy that data wherever you sort of kept the data in HDFS maybe - to the - to the message bus again and then run the same topology could encode on yesterday's data as well you know what that's really inefficient right that's basically you know that's that has very very heavy network penalty the other option obviously is that this is this is another kind of store but then it kind of boils down to a kind of a lambda architectures anyway right so you know a copper really you know reprocessing logic kind of you know fails that and reprocessing happens all the time and we will show that with an example towards the end of a session of why reprocessing is very very important happens pretty much every day and companies like Twitter so you know just summarizing this patchwork of systems written at different places maintained by different people so you need you need actually a team who is well-versed in each of this technologies to be actually able to just barely run and and there are too many places where things can go wrong right too many failure modes and and when the API is when the semantics are not clear between the messaging and and compute and the storage and so on so for there are lot of losses possible that are hidden you'll only notice once you know your business logic or your business results go and match the expected what your client expects and so on so forth and then you'll probably be in for a nasty surprise in some sense so what what people really need here right I mean at the end of the day what people need is no data is coming in right they want to be able to a act on the data you know if you know data is coming in store the data as fast as possible there are probably a bunch of subscribers that are waiting for the data to do some action right so you need you need something which basically says okay what there's a bunch of subscribers I will give the data as it arrives as fast as possible in a scalable manner to all my subscribers right and that's usually what's called the messaging you know stuff right and then you want obviously want the data to be persisted right you never want to throw away your data because your data you'll you will reanalyze it yes tomorrow you'll realize this next month and so on so you want you want to like a linearly horizontally kind of scalable durable storage right and this this is usually you know you need you need make sure that the data is replicated to avoid data losses make sure that if machines die the data is automatically pop you know replicated and all the stuff that you sort of taken care take for granted with the distributed file system kind of kind of thing you know obviously you want like high availability and re throughput and so on so forth right and at the end of the day the processing is very similar it's like a snag you know you want some way in which you can sort of express your competition as a dad with some input source and some and some you know dag operators to basically operate your data right let's come to the messaging and the storage aspect of it right I mean in the face of things you I can actually say you know what messaging and storage are probably two different things right you know in the sense that you know one is actually waiting for future data the data that will come in or it has just arrived and and it is very very low latency you know you need to be able to kind of quickly give it to the subscribers and move on right storage and the other hand has the duopoly in the sense that you know what you're actually storing past data and here the throughput you know matters more than than the latency so it seems that you know what they they they they are talking they are having two different things but but but the fundamental thing is that they are actually the same data right you know the data that you come in is the same data that you're going to store in your messaging in your storage systems so why keep it at multiple places why not keep it at the same place we will come to some of the challenges some of the differences there but ultimately you are dealing with the same data right now what if you've got in one system something where you know a very fast you know pops up are distributed message system where in the data comes in and the data is is able to be immediately sent to all the subscribers at very low latency at the same time the data is not thrown away right the data is actually backed durably by us a very horizontally scalable lock store that lock store that does all the stuff that you that you know take this distributed file system for granted right what if you've got this in just one system one place where the data kind of kind of stays right if you have if you've gotten that your system your architecture becomes inherently very very simpler right you have a single place where the data resides and therefore now you have the ability to basically store and compute on that data whether it is today's data or yesterday time in the same system right and and and the ramifications for the computes are even better right and now you know if if you since you're talking about data coming from the same system you know the same stack that is actually going to work for today's data or right now's data will actually work for for yesterday's data right one of the fundamental disconnect that is there in the market is is is traditional distributed file systems are not not giving up with are not giving that benefit of low latency they are too much into the high throughput mode and that's the reason for all the dichotomy that sort so comes up you know with multiple systems and so on and so forth but what if you've got this the answer is we actually have this already today and that's you know pulse our bookkeeper and to talk about more on pulsar and bookkeeper and why that that thing makes sense how that is activated I'll ask CJ to sort of come on board so as Sanjeev said is the whole point is the domestic data coming from real time is actually became the historical data store in a in a storage so what if a system they provide single storage obstruction for both real-time data in in history of data so then you can use the same unified interface to access in both high schoo data in the real time data so that's what we came into the picture is y pasa has the pasa the combination of pasa in Pookie has the value to bring in the Unified Messaging storage story so to talk about a bit about pasa at ease so what simple sentence to describe what is pasa pasa is a fast durable in multi-tenant pub/sub messaging and it opens us by Yahoo last year around October it's now in Apache Incubator for incubating and approach pasa is build of in a layer architecture what does the mean is it's comprised with its separated into two layers one is serving or we call messaging or serving and the add layer is storage is where data can be keep for a long longer time and this is the separation between the serving in a storage king can bring in a bunch of the benefits for example like independent scalability that means if you want to scaling your messaging are serving there you can just drop in as many brokers in a serving there if you want to keep data for much longer time you can just simply drop in the as many number as many number of bookies as the storage and there's no need for data dependent thing you don't need to say I want to expand my cluster I need to do expensive data copy it just automatically shifting the traffic when you drop in the storage node and because it's a steady serving that means your messaging there is mostly studies studies the benefits of stanley's is when you want to shifting the traffic for online service or for any instant it even quickly shifting the traffic between brokers without expensive copying the data and that is the key to multiple things like multi-tenancy and low latency iOS elation that is all the benefits and if you can see here the in the layer architecture is you have two layers dealing with different aspects is for the serving there you can deal dealing with master messaging dispatching low latency deliver on a storage there you focus on storing data for my longer time and keep the data not the node a Halasz and durable stuff so and the knees of the pasta is using buquebus a scalable lock storage and bookkeeper itself is we call it it's a duplicate lock storage and it provides bunch of the benefits is its low latency durable right that means by durable here is not as replication it's also mean durably persists the data on disk before acknowledged that means all the data would be on disk rather in filesystem memory are in cash so this is critical for if you want to use and this is critical to Unified Messaging and storage system because if you want to keep data for much longer time you want make sure the data is to be stored on disk without no data loss and it also provides a dog provides a very simple repeatable read consistency so no matter how you want to process data you can cut exactly the same sequence of data that is important for data reprocessing because you don't want to get into computing engine that when I process data yesterday's data again and again I got different reach out then that is not good for processing engine and it's also built in as a bunch of other features like high availability and be able to store many locks on the storage now so it has the ability to scale the number of the locks to minions or even tens of minions and that is critical for like things like micro service you will you might have a lot of small micro service you want to consuming from large number of the concurrent locks or topics and also iOS elation that is the key to the success of multi-tenancy in also low latency and for more information fill up too check out the the project and because of the layer architecture it brings a fundamental difference with other locks the lock system are like pops up system is it became more segments century what does that mean is segments are Prague is more coming from file system our storage perspective you break down large chunk of data into segments and you are able to dispute other segments across the storage node and that is what on the right right hand side is you can see how a poster topic a parcel extreme can break down into small segments in dispute into storage node and that on the left side is show just show an example what is the difference with other system that cop car which is partition century and because of the difference of the architecture it brings a lot of benefits in this segment century system is for example if you want to add more cap city you want to expand your cluster you can just drop in new node because its segments entry so new data I will just quickly be breaking into new segment and going to new storage node without rebellion seeing the partition without expensive copy and also if there's any node failure you you just need to repair the broken segment you don't need to recopy the whole partition data in order to recover and there are also punch of the benefits of segment century architecture feel free to talk to me after the talk and beside these different there of the architecture one thing we we found interesting in pasa is in the mastering mode people typically categorize the use case into two categories one is streaming and one is queuing this typically patan in different systems like for example queueing people my using web MQ a lot in the in the market for streaming people may use Kafka a lot but there's no fundamental difference about streaming and queuing they are mostly for propagating all the real-time data ambassadors this unification in the meshing model through a topic in subscription model so in a given topic you can have different type of subscription and a subscription define the way how you want to consuming the data for example on a tab is exclusive that means means you can there's only one consumer it can read from one topic or one partition that is doing the sequential read and exclusively and fare over means you have multiple consumer one to consume from the same partition but there's only one active the Alice became standby when the consumer fell it would just be the other consumer just take the over for four people woman familiar with Kafka the failover mode is what it's called a petition dependency in our consumer group in Kafka so that is for streaming and for queuing that is basically implement in as a share consuming that what does limine is if you have a given petition in exclusive mode you only have one consumer to consume so how fast you can consume the is limited by that the consumption capability but if you do a lot of expensive work like for example like video encoding image transformation or maybe some very expensive or complicated event processing if you want to scale the capability you in you don't care about authoring you just need to add as many consumers as you can because they're they're consuming their share sharing consuming from the same topic so this is output in into one messaging model that means you don't need to you don't need to distinguish like how people want to define different consumption behavior they the consumer can scale on their own way and besides that the posts are provide not just a Fed topic namespace it's more hierarchical what does that mean is when you embark pasa you on bought as a tenant each tenant would have a property and under the property you can create multiple namespace in their space you create topics so in this way you organize a topics in a hierarchical way that is useful for in in a company that you might have different teams different in my use different topic processing different business logic so in this way you can in segments organize your topics and you can grant the some kind of access control so for people they may only care some of the topics so critical data they are not be visible to other peoples so that is the way for people to achieve multi-tenancy and besides there's the other important things for mastering system is you should have capital capability to replicate data between data center for the this data center it is a disaster recovery so this is an imposter in a very nice way is it's pure in in the messaging there which is in a broker so when you want to enable cost sorry duplication across data center duplication it you just config forgiven topic which cost or which data center you want to replicate who you don't need to set up additional complicate you know maker to run a additional job what does it mean to a lot of use case is if you have a lot of data center they were simplify your management if you are running in IOT area you might want to edge to cloud replication and it's so hard to managing a lot of Ages if you using additional job so this these purification facility bring in a lot of value to to simplify your architecture and Kafka aside posar provide a lot of kind library like Java C++ on in WebSocket for people to access and it also supports petition topic so you can still scale the the topic with increasing the number of petitions and it also for people who want to familiar with Kafka it also provide a backward compatible API for accessing the data so it's just additional to existing parser library and it also does a bunch of things like trend transparent batching compression tiers tier as encryption in authenticate authentication in that the other thing one caught out here is it also provides end-to-end encryption what does that mean is for a given producer use you have the ability to encrypt data before you sending to a poker and broker so the storage system so start these encrypted master it transparently so they don't know how to decrypt and only consumer be able to decrypt the system so that is important for a lot of use case that require a high security demand and that is a overview about how pasa and also the Pookie bill looks like and how what is the idea around unification on mastering and storage in the hand to send you've talked about compute side so you know Helen incidentally was built right here at Twitter and you know a I'll just quickly go over this and probably try to spend more time in the examples because that's where this thing hopefully will become more concrete you know it's just a processing dag right so directly a cyclic graph and a bunch of sources bunch of you know you can do a bunch of transformations and so on so forth much of you guys are already familiar with the concept of sort of processing dag right now one of the key things is that you know Helen is actually just a library it's not a system so there is no Heron cluster running everywhere right one of the key decisions we sort of made when we're sort of designing Heron was that you know what nobody wants to maintain another Nimbus equivalent right it's it's it's too complicated you know there's already one that only maintaining and that's the scheduler right you know whether it's goober net is me so she on or whatever most people are and increasingly even mid and small-sized companies are also having you know a scheduler of their own increasingly kubernetes so what it means is that the scheduler is the one that actually kind of manages your resources and make sure that you know things are allocated to the right ones appropriately and if things go bad can I restart some and so on so forth so all that logic kind of thing that was there and the storm kind of went away all that thing went away when you sort of dealt with design heron so having just a library so you know you submit your topology and the topology essentially act as a job running inside a scheduler so imagine if you--if you're running it as as as a resource in the me source scheduler it's it's just a regular resource job it it has a bunch of containers inside inside the job it's at the end of the day from a scheduler perspective it's just a job right and and that's that's one of the reasons why you know there is no system to maintain and that also makes the thing very very scalable right there is no single point of failure or single point of bottleneck for a heron topology all the topologies are acting as independent jobs working of their of their own one of the main things we want to do is task isolation right you know think doing disparate things in multiple threads in a single process sounds good at the beginning but when you are really dealing with high throughput data and so on so forth it's just a bad idea and these days with with processing power and memory being so cheap which actually makes sense to forego a little bit more in terms of memory and actually go for task isolation as as as in all the tasks running as individual processes right so if you look at the if you look at a particular header in topology you know in this case there are like no three containers for this particular topology each of these data containers are the instances I won I - they're actually running as processes and these could be your sources your sinks your transformation bolts are and so on so forth and in addition there are a bunch of other tasks running what this actually means is that all of the tasks are isolated by their own and they actually have process level isolation not just thread level isolation which means that they can actually be more manually controlled more manually shoot double shoot several shoot right and the other thing is that these instances are written in their native language so heaven actually supports a bunch of languages C++ Java and Python so when you're running a Java topology there is only Java in your stack there is no j'ni happening anywhere in the inner thing so when you actually get when you actually get an exception it's a pure Java stack so pure Java thing rather than a mumbo jumbo of different layered stack differently languages it actually makes developer productivity increase tremendously because now you're dealing with you're dealing with a particularly egde runtime environment which a developer is probably very well versed with right so if you're running a Python topology for instance the instances are all Python processes rather than any kind of you know things just invoking Java and so on and so forth this means that any kind of Python processing any kind of Python developers who's familiar with the Python debugging tools and so on so forth can now attach the debugging tools right into each of those instances and can debug them as if he or she were doing in a regular Python development well the other interesting thing is the backpressure of where you know we don't drop top also when we're you know in a processing dag some nodes are fast some nodes or not fast some processing logic is inherently expensive and and if if particular downstream components are not able to keep up you have the ability of saying to the source you know what we cannot consume let's stop consume for a while until the traffic connect leads up at the downstream components and once you sort of have the clear-up component you can you can then start consume again this essentially means there are less data losses in the system and this actually has resulted in a tremendously improved latency and accuracy as well and you know usual you have multiple layers you have the you have the old you know stomm compatibility layer where if you have a storm topology people can actually program in spouts and bowls and so on so forth which gives a little more control but probably a little more verbose to program you also have the streamlet api which is more funky so you have your map flatmap kind of equivalent and you know sequel is in the works oh you also will have a declarative way of expressing your computation and so on so forth so multiple different ways of computing the same things are available and here is where I want to hopefully spend a little more more time you know just take a very simple in a use case right you know you you you've analyzed your data you've created a classification model and what you want to do is as soon as a tweet sort of comes in you want to classify it in in some sense using that model and and once you sort of classify that you want to be able to sort of add additional attributes to the team to the street to that particular tweet and so on so alright this can easily be constructed as a real-time real-time topology right now what's actually happening is that when your real-time topology is running you know this is how typically it is actually you know running you know you have you have pow you have the sources you know you have a the source is a tweet stream that just say it has two partitions in this case and and you are your your real time job is actually processing those incoming incoming tuples so here is here is you know to source instances each acting on partition 1 and partition 2 and they are actually part of a processing of offset offset there right this is a very standard way of doing a real-time computation job nothing nothing special here right this is how this is how you know real-time competition actually works but and here's the interesting thing right frequently usually almost always what happens is that you know tomorrow you have a better classification model right now you you you you you've you've ran some more data and this classification model you think has more models more classifiers and is in general a better accuracy in whatever regards you want now in addition to actually classifying your incoming today's you know tomorrow's incoming tweets with the new classification model you actually want to classify yesterday's tweets as well or for that matter last month to each as well because you know you want to be able to you know now that you have a better model you want to be able to classify all your tweets and have some kind of extra time or whatever it is that you want to do right without unification strategy how would you do this it's it's actually it's actually very cumbersome right and if you if you were in the lambda architecture what you would actually do is that you would go write a spark job or or a MapReduce or or have a higher level like something good run run those things for you right so those programs are now running in a different stack yeah and because they are running in different styles there's a lot of tuning and so on so forth to be sort of to be sort of made and so on so forth right with the cop architecture what you need to do is that you need to bring back yesterday's tweets our last month's tweets back into your your your messaging pipeline and then run other topology as it is right in either case very cumbersome either network or developer productivity like really hurts with it with unification right we we sort of said how distributed log how pulser and and bookkeeper kind of unified that storage and messaging layer you don't need to do actually anything right these were your partitions of the tweeds and and they they actually were naturally stored at segments and all of them are available at bookkeeper already so all you need to do is that when you start your job reprocessing yesterday's tweets you just have to choose a different offset and and that's about it the system does not have any other limitations besides that you know you you just start the processing of that and so reprocessing now is very simple it's just relaunching that job with a different offset and things automatically sort of you know get get sort of taken care of off by you right that's just one example of how you know reprocessing with the unified architecture kind of makes it makes certain things which had a lot of high friction very straightforward unfortunately we are kind of you know out of time you know if you want to learn more obviously there's a lot of blocks and other stuff on our website as well as in each of these you know a project website so feel free to go and browse any questions would be we'd love to take either now or later yeah so just just be clear with me with this kind of manta style approach a lot of people say that that's maybe not such a great idea in the long run and I think you're talking about the end there this notion of like you just like you know we don't run the job sir with the offsets but what's the story for like you know machine learning we often want to go back and do like backfills and some some companies on my footer have smaller data then they can actually go back and read back all the time they have you know just like you know ten years worth of data and coccyx you just go to point the job back you know arbitrarily back so this is that the vision that you see like with on the back kind of what are your thoughts about the long-term like breaking transactional kind of like [Music] that that's that's essentially our vision right so a couple of points that you raised right like nobody's doing that with Kafka of actually pointing it to last year's worth of data and running it because by definition remember the Kafka partitions are limited by the disk capacity on one particular broker which means that they all have a certain finite size now because keeper and pulse are breaks a partition into segments and these segments are then distributed across your cluster there is no real limit to the size or the length at which a particular partition can grow right which means that you can keep of very long partition that spans across an entire year for instance right all of that now makes exactly what you said possible when you just point out to the right offset and the same thing basically same thing basically runs that's exactly our vision and that's that's only possible if you have a segment you know architecture segmented we have architecture in your partition rather than just a partition sending architecture you hit it right that I think is what increasingly people will start will start doing [Applause]