Scale By The Bay 2018: Sherin Thomas and Micah Wylde, Stream Processing at Lyft with Flink and Beam
Laika we are both software engineers in the streaming platform team at lyft today we will be talking about how we are solving some real time processing challenges using stream processing so this is okay so this is a rough agenda for our talk today so we will start with some some use cases and then we will talk about some concepts of stream processing and also some of the interesting challenges that we face when setting up stream processing programs and how we are solving those so let's start with some use cases so I hope most of you have used lift at some point or the other we are a right sharing company our goal is to match passengers with drivers and we want our drivers and passengers to be really happy with our platform and that means short ETA s we want to make sure our drivers are being paid fairly while at the same time making sure that our passengers are not being overcharged for their rides and we also want to make sure that our users have a good experience on the platform and this might mean timely notifications and so on to achieve any of these use cases we make a lot of decisions in real-time for example in case of pricing we have dynamic pricing and we take into account real-time supply demand curves and etas and so on and these change a lot based on traffic and many other things so we want to make sure that we are making these decisions in real time we also want to combat fraud on the platform because there can be real monetary consequences and we want to make sure our users are getting notified if their right is getting delayed or if there's an 80 a mismatch any of those on the driver side of things we want to make sure our drivers are earning well on our platform and also getting more rights this means that we want to make sure that they are directed to places where there is a high supply for example or we want to make sure they're getting their bonuses fairly so some of our bonus structures involve like for example counting right streaks so depending on whether they the number of rides they completed in the last maybe a day or a week they get bonuses so we want to make sure these are counted accurately and bonuses are fair plate wealth so for it to solve all in any of these problems we generate a lot of events on our platform and we use stream processing to do these computations in real time and Michael will be talking yeah so I should mention real time data is really really important when you're a real-time business like lyft where people you know expect to pull out their phone and be able to get a ride immediately so to kind of talk through kind of what we mean by stream processing what these systems look like I'm gonna start where most of us started with data processing these are batch processing systems so the basic idea here is we have some some system producing events a case of lift this is our mobile apps both on the passenger and the driver's side which as a user pulls out their phone and starts looking at destinations eventually books a ride all of these things are producing events back to our system similarly as a driver drives around the city we're getting continual pings from their phone telling us exactly where they are so that we can match them with riders so as all this data is coming into our system it goes to some data store data Lake HDFS s3 HBase and then periodically users can query it using a system like MapReduce or presto or hive maybe you have some automated system that's wearing it every day and generating features for your ml pipelines or whatever and this works well for a lot of use cases it's very scalable up to petabytes of data and it's fairly straightforward for users to use but for a lot of use cases like ours it's just too slow you know if there's an accident on the highway and it takes us a day to learn about it and to update our ETA systems you know that's just too slow we can't tell a user that their car is coming in five minutes when there's a giant flaming truck on the highway so so lift realized about a year ago that we needed to be faster and a lot of the things we we do so we started turning to a stream across the scene so stream processing is a different way to compute things on data where instead of kind of collecting a whole lot of data and then every hour or every day doing some computation on it instead as the data streams in we do computation on every single event for every event that comes in in you know real-time we are updating our current state of the world so for lyft this might mean deciding how many drivers and passengers there are in a particular part of the city so we can decide whether we're gonna turn on primetime or not it might be deciding that a user is trying to defraud our system and we should block their account all this stuff happens immediately and the the time scale we're talking about here is is often milliseconds or seconds so it's stream processing requires thinking about your computations a little bit differently than you might be used to for batch so I'm just gonna go through a few of these concepts that that come in when you start moving towards a stream processing so first of all you're probably gonna want to divide up your data into time in some way there's a few different ways people do this the simplest to understand is a fixed window where we have say you know data from 5 o'clock to 501 501 to 502 and we do some computation over each of those windows slightly more elaborate it's a sliding window where we still have a fixed length window like say five minutes but the window slides by say a minute so that we can get some notion of history while still updating it on a regular basis and then we get to more interesting things like session windows where we're connecting all the activity for a particular user so for example everything that happens when a user pulls out their phone selects the destination or as a lift a session window could connect all of that and do some computation on the the whole kind of user session so these aren't like nice high-level concepts that allow us to like simply express some of the the things that like application developers might want like you know an ml engineer might say I want to know how many cars there are in this part of the city in the last five minutes and that's a nice like high-level thing to be able to express but there's a lot of challenges and actually computing that in particular in the streaming system like data is delayed it's out of order this particular challenge for the lift because like our data comes from our mobile apps and mobile apps are on slow cellular connections they might be buffered by the app they might be offline for a time period so you know we might get events that are arbitrarily delayed but we still need to be able to to do these computations in a timely way so the solution that people have come up with for this problem is to introduce something called a watermark and to explain what that is we first have to kind of formalized time in a stream processing application so we have to have two different notions of time here we have the event time which is the time the event actually occurred and then we have processing time which is the time that our system saw the event ideally you know these would be basically the same it was maybe some delay but in practice you can see this reality line kind of zigs and zags and kind of be all over the place so in order to like not wait forever to do our computations because we don't know when the data we need say for our 5 p.m. to 501 window we don't know when that data will be there but we still need to decide to like do our computation at some point so modern stream processing systems introduced something called a watermark which basically tells us when we think all the data for a particular time period has arrived so you think of a watermark as an estimate that at a given time we've received all the data prior to that time so visualize this a little bit better the Google data flow team put together this great visualization that shows what happens as a watermark kind of moves through time so here we're doing the integer sum over events for a two minute fixed window and as these events come in you can see with the the processing time bar we're summing them up then we have this this green water mark that we're estimating when we've received all of the events once that gets to the end of the window we can close out the window and compute the the data of course data might still be late right we have this random 9 point that comes in you know a minute later and then we have to handle that in some way there's basically two strategies you either just drop the data if you've already used the computation or you can try to update the existing window and and kind of read out put that to whatever system is consuming this so I lift to actually write these stream processing applications we've settled on Apache flink which is a general data processing system Sharon's going to talk a little bit more about so I lived we are using three Apache flame for stream processing so what is blink blink is a framework that allows stateful computation on bounded as well as unbounded streams of day it can scale to millions of events per second and also terabytes of state can be managed flink also supports event time processing and this is really important for our use cases as micro mentioned a lot of events are generated on mobile platforms and they are batched and sent to our event ingestion pipelines fault tolerance and exactly one semantics are also supported which is and this is made possible by flings checkpointing mechanism checkpoint is essentially a snapshot of a state of a fling program at any point and that also includes a position in the if in the input stream that is being read so this is a simple word count example for for fling in Scala and what this is doing is it's streaming in a text and counting the number of words in a window of over a window of five seconds plink also supports specifying computations or aggregation in terms of sequel so in this example a data stream is being read as a table and the aggregation is specified in terms of the sequel query and this is a lot of programs at at left uses this fling sequel so at lift we have right now we have more than 50 fling jobs running in production tens of millions of records are being processed per second it's currently deployed in our ec2 instances using our custom tooling and custom deployment methods we have also been working on getting flanked on kubernetes that work is in progress most of our future generation for ML models are powered by Flynn sequel another thing about like lift engineers so at lift Python is the primary language that is being used flink currently supports api is in scala java sequel so in order so we don't want our engineers to have to learn Java to right-wing programs so for this we have been working on supporting beam at lift and beam is a is basically a unified programming model that can that can be used to write batch and stream programs and flink as a runner is supported and SDKs and Python is available for example and also it supports it has support for complex been doing and triggering methods so now let's talk about some of the challenges like interesting challenges that we faced and how we solve them so first I want to talk a little bit about bootstrapping so what exactly is bootstrapping so for that let's look at an example so this is a fling sequel program that that's counting the number of writes canceled per passenger over a period of seven days so the window here is seven days so what will happen if we start this program today and and it starts consuming live events for this program to be able to answer the query correctly it needs to run for a period of seven days process that data in order to answer the query correctly and waiting is the hardest part what happens if after seven days we realized that we needed a change in the business logic so we can't really like start it now and start building the state from today how about total cancel since the beginning of lift four this will have to go back in time and start this program the day I left started so this is where bootstrapping comes in so instead of building the state today using live events what we do is we read historic data to build the state and bootstrap the program and then we are obtaining the state as live live events come in so now our program returns results on day one at lift historic data is in s3 and live events come in through Kinesis pipelines but Kinesis has a retention period of seven days so if the window is longer than seven days and we want to process historic data we cannot rely on like just Kinesis so we also have to read events from s3 and there were two things we wanted - there were two requirements we had when we were building this so first is we want bootstrapping to be fast so when the program is bootstrapping a bulk of data is being processed so it can it can be a slow process but we were willing to add more resources to make it go fast once bootstrapping is over and the program transitions to the steady-state and start processing live events the amount of data that is being processed obviously reduces and at this point we want to reduce the resources to save a safe cost so this is the first iteration of the solution that we came up with so what we did was we essentially had two separate programs for bootstrapping as well as and a steady state so we would first run a high provision job that would read historic data from s3 perform apply a business logic on it write to the sink once we know that we have finished processing historic events and we know this based on water maps we take a snapshot of this program cancel this program now we start a low provision job using this snapshot that we just created and this program would read from Kinesis and process live events apply the same business logic right to sink and this worked for a lot of most of our cases but there was a disadvantage so in fling in order to apply a snapshot the requirement is that operator you IDs should not change so since we had these two separate programs over here so the flink allows setting you IDs manually but a lot of our programs use fling sequels so the operators had Auto generate new IDs and secondly if the shape of the program changes then it's harder to reapply the snapshot so this method was really clunky and not all our problems are solved so this is the solution that we came up with next so what we did was instead of having these two separate programs we combined the two so now this job reads old historic data from s3 it reads live events from Kinesis unions the two applies business logic and writes to sync once we know that historic data has been processed and we determined this on the basis of a watermark so what so here that green line is the watermark that's moving forward so when we know when the water mop moves past the target time we know the old historic data has been processed so at this point we take a snapshot of this program cancel it and restarted or restart a much lower provision job using the snapshot so when our program starts it's a higher provision job with high parallelism but after bootstrapping is over we use the snapshot to restart it with a lower parallelism and this worked really well like right now we are using this method in production an advantage is that it's a single program it's seamless we have external automation to detect bootstrapping based on watermark and we are saving on resources however the disadvantage is that when since we are unioning data from both so Kinesis the live events coming through Kinesis are not being used until all the historic data has been processed and this data was being buffered and flay this this is a limitation that we'll be talking about next so because of this data being buffered in case of like programs having large windows this was causing performance degradation and this is a common problem in any scenario where there is a source queue so we started thinking about this and came up with a solution and Micah will be talking about it next cool so source Q what does this mean so let's say that we have a partition data source like cough carrot Kinesis and each of these partitions has you know a collection of events that are getting streamed in and these events come from you know different parts of event time so in a well-behaved system you know the oldest events will be at front the newest events will be you know at the back and all the partitions will be basically in sync together so here all the events were reading off of our partitions these little circles are in our like first window so here the little job that is counting just the number of events in a one minute fixed window so all these events are coming off of our first window and then you know we keep reading events and all the events now are coming off of our second window which means that we're able to close out that first window we're able to reduce all of the events to just their sum and throw away the raw data begin right and that's great that means that we're able to output our data to our our consumers and kind of be done with that period of time what happens of our partitions are skewed if one of our petitions has say a bunch more older data than our others in this case we we start reading in our first window and everything's good and then we start reading in our second window but we still have data for the first window that's streaming in which means that we won't have gotten our watermark yet and we can't close out that window so as we keep getting more data in for newer time windows these old windows stay around with all of their raw data this is really bad in the case that Jaron talked about where some of our data is coming from a system like s3 it represents historical data so since all this data is is streaming in while we still have the the live data coming in we have all of this state that's building up in our application this causes a bunch of bad effects in flink checkpoints start taking a really long time because we now have to checkpoint all of this data that's been buffered up if you're using an in-memory state store you might actually just run out of memory and if you're using a dispatch store it's gonna be a lot of flushing this data to disk and reading it back in so this is really really bad and has caused a lot of outages and other performance problems at lift so this became a really high priority for us to fix and I want to talk through how we how we addressed this so let's look at this kind of from a systems perspective a flink application is made up of a bunch of nodes and inside each node we have some threads that are consuming off of some different Kinesis partitions the way it works in kind of the open source Kinesis reader each thread or each partition has a single thread that's reading off of that that partition and then writing the data to the rest of the system so the first thing we did was introduce a bounded priority queue after the reader thread so now the thread reads data off of the partition it writes it to this queue and then we have another thread that is choosing the oldest element off of any of these queues and sending just that element to the rest of the system so this means is that if one of our partitions become skewed and has newer data it won't be read from so it just starts building up until it fills up entirely and then the thread that's reading off of Kinesis can just stop reading from Kinesis this is great this means that our data that we're not able to process yet because this chinnu is staying in our pub subsystem which has basically infinite retention instead of being streamed into our application where we have to start a memory and check one and do all this other business then you know once our other partition catches up our sender thread starts reading from both cues again and we're able to start consuming from the partition so zooming out a little bit more of us you know in a real funk application we have multiple consumers so we need some way to coordinate them using some shared state so we have an arcade zookeeper that is maintaining a notion of a global watermark which is basically the minimum watermark we've seen across any of our consumers so now as as our threads are pulling data off of their internal cues they're periodically also updating this global watermark so they update the watermark that gets replicated to the other nodes and then the other nodes know to only consume and produce if the data that they're seeing is within some bound of that global watermark so here our current global watermark is it's the oldest time so if our second consumer starts seeing newer data it's just going to stop producing to the system which means that his cues will back up and we'll stop consuming from our pup sub layer and then first once our other consumer catches up it'll update the watermark which will update the other consumers and then they can start consuming again so current state of this we've built it in-house and we're running it we're also working on contributing it back to the open source community to be fair this is a common problem that a lot of companies are facing so for interested in that follow that flick JIRA and that's all we have for you so thank you so much and I think we have a few minutes for questions [Applause] talk about the change in the programming model after you did this whether it change to how you write in how we implement this and then all those training versus combined pageantry sorry I submit the bootstrapping not really it's the same flink program so so bootstrapping here so we are running the bootstrapping program that's also a streaming program and we did think about using batch batching or like using the batch model for stream for bootstrapping but the issue there was that there was no easy way to use the snapshot that was just created by a batch program in a streaming program so that that was mainly the issue so we did consider like either using batching for that the other alternative that we considered was using a tiered storage so either like like I think right now I'm aware of Apache Pulsar that has a tier of storage so in that case we wouldn't need like two different sources that we can read from but yeah so does that answer your question yes yeah so artpop subsystems are so art pipe so we are sort of provisioning Kafka right now for our pub sub for like for the event pipelines and so on but flink is mainly used for processing like for any place where we need been doing an aggregation and where so for us like event time processing and exactly once semantics was important so that was that was like two main criteria that went into that decision yeah I yeah yeah so spark has recently introduced structured streaming and they're starting to get their latency down it wasn't really an option when we made this decision but I'm definitely excited to see that mature oh yeah and also we're heavily adopting beam so one of the big benefits of beam is that lets you write your job once whatever language and executed on any execution system so hopefully they'll give us a portability in the future if we decide we want to use different execution systems yeah so in I in iteration one it was actually two different programs so there was one program that would just read like that had just one source which was an s3 source and the other one Achilles the source in the other one we we have a source that unions the data from both so from a user's perspective they just run this one fling program and magically it will just transition from bootstrap well magically like we have external automation for that but yeah so one of our largest programs is one that we have for fraud one of a fraud mod that Pao is one of our fraud models and what it does is it looks at last 60 days worth of client actions on the app and client actions like the highest volume event and for this bootstrapping took about four hours so you said that you started using [Music] no so our main motivation for adopting beam is actually to get python support because most of the developers have lived our Python developers the data scientists like Python so that's Python on flink is super new were actively developing it along with Google and so that should be sort of production ready in the next like three to six months we're already starting to kind of roll out use cases internally for that yeah I just went to the geo talk and I thought that was really interesting definitely interested in like learning more about that so you can [Music] now if you just have one partition then there wouldn't be a skew problem but I mean at our scales you always have lots of partitions and naturally multiple data sources which which really exacerbates those key problems cool there's no more questions thank you all for coming and we are hiring so please reach out to one of us if you're interested in working on some of these problems [Applause]