data.bythebay.io: Prakash Chockalingam, Deep dive and best practices of Spark streaming
Recording: data.bythebay.io: Prakash Chockalingam, Deep dive and best practices of Spark streaming
thanks fall I had a whole slide on that thanks for so just a brief introduction about data bricks so I'm a Solutions Architect data breaks so I work with our customers on solving spark related issues so just a brief background before the talk the airbrakes is founded by the creators of spark and we provide a cloud enterprise data platform which has which comes with manage spark clusters and it has notebooks for doing interactive data signs and then running production jobs all with like enterprise-grade security so the reason as today's talk is about spark streaming the best practices so this comes from my experience working with our customers our making them run sparks streaming and production so the agenda is like give a brief introduction to spark streaming how it works and then go a little bit in depth about a lifecycle of a spark streaming app and then so that should set the context for the remaining two sections which is the aggregations and the state management that that can be done on sparks streaming and the best practices for doing them and finally operationalization tips and key benefits for of using spark streaming so what is Fox trimming so sparks serving provides a high level API on top of sparks rdd's to do joins and window based aggregations so it's mostly like it has similar interface like rdd's the functional so it follows the functional programming paradigm and it guarantees exactly one semantics even for stateful operations and you can integrate that with other libraries on top of spark ml lab or data frames or sequel so in terms of people using it there are more than 500 companies that are using it in production some of the major ones I've used that when I was at Netflix so getting to a cold part of how it works in a nutshell there are data streams coming in so there is a receiver that sparks streaming kind of uses it to receive data continuously and then it chops them into batches sends it to the spark or engine and then pushes out the results to the sink so that's in a nutshell at a very high level I'll just walk through a word count example so that it sets the context for then subsequent sections so any entry point for a streaming job is the streaming context and you can create a streaming context by providing a a batch interval as the parameter so batch interval is the the unit of time that spark spark streaming uses to chop the data into batches basically so once you have a context now you can provide a receiver or you can specify the source information to say where the stream is coming from so in this case it's a simple socket text stream and what you get back is the D streams or D stream is the API that spark streaming exposes to kind of do any transformations so once you have a D stream which represents a data stream you can do further transformations on them to create new newer D streams so in this case we got a lines D stream so we are doing a flat map operation which kind of splits all the input word input text into words and you have a words D stream and then further we can kind of do for the transformations and do a final word count get a final word count stream and then here we are just printing the D stream contents onto the screen so until now like nothing happens once we do a streaming context at start so that's when like the streaming job kicks in like Spock goes ahead and sets up everything on the cluster and the streaming job starts to run so let's get on to how that works in order to kind of have an understanding about some of the best practices that we can follow while doing a streaming job so this is an execution model in any spark application user code runs on the driver as soon as there is like a distributed collection like RTD and there is transformation on it spark kind of pushes those operations to the executors and does them parallely and as far as a resource manager you people normally use yarn or me source or Spock standalone so this is a execution model for any spark application this is slightly it's slightly different for a streaming application so for a streaming application the driver code starts execute execution on the driver and as soon as there is a line for a kind of receiver so spark kind of chooses one of the executors and starts running a long-running thread on one of the cores in that executor so here you can see there is a receiver thread that gets started here and that starts pulling in the data stream from its source from wherever it is and it starts storing them as data blocks in the executor and if required based on the you can provide a parameter to spill it to disk if you think it won't fit into memory and you can also specify a replication factor so that it kind of moves the data blocks to other nodes in case of a receiver failure so that's the first part of a spark streaming application where it first sets the receivers the second part is the actual execution of all the operations and the actions so as soon as so the operations and actions are basically the word count the actual word count doing a flat map doing a reduce by key and kind of doing all the logic to do the word-count so based on the batch interval provided so here like we provide if we provide a batch interval of one second Spock streaming kind of fix that for every batch interval it submits the jobs to the executors where the data is located like just to have data locality it executes the tasks and then pushes out the data to the sink just to go one level deeper so this is a streaming application which kind of programmers right this is an example here which has two socket streams and then there is a union of the two socket streams followed by three actions the first one is basically saving it as Hadoop files the second one is doing a map followed by a for each so basically it kind of for each is any action you can like do a parent or push it to some other sync similarly there is a filter followed by a for each so what happens here is this program first gets compiled to addy stream graph so here there are two source two receivers followed by the Union followed by a common operation and followed by the three grey ones which which indicate the final output operations so once this is there the next step is like for every batch interval spa our spock streaming will take this dag and starts putting it into a queue so the queue kind of you can think of rather like a simple queue where all the tags are queued up for further execution so this is the role of a Spock streaming the core of Spock streaming the next phase is where SPARC picks up from this queue the first day the first dag that it needs to execute it breaks them into stages like wherever there are shuffle boundaries like it breaks them into stages and then it breaks the stages into further granular tasks and then pushes it to the executors so the executors is where the actual the data is being received and the data recites so this all this happens on the driver for every like a batch interval if the batch interval is one second then for every one second like this dag gets compiled it put it is pushed to the queue and then spark picks from the queue compiles the stages and then breaks them into tasks and pushes it to the executor so this is like you know end to end view of a spark streaming app so in terms of like aggregations so we saw a word count so word count the first program what it did was a word count for every one second that's not a real-world application in any sense so the most common thing is doing word count over a time window like for the last 10 minutes or the last five minutes or even for the last one hour so here each square basically represents a batch and the dark green basically represents the entire window size that you want to do the word count for and the lighter green represents the sliding interval so it's basically like a tumbling window so you can do a word count for every N seconds instead of doing for every second so the best way to do that would be like you have addy stream and then you can say instead of just reduced by key you can also specify a window parameter and then specify the function the aggregation function in this case it's a simple sum followed by the window size and then the sliding interval so this is great like it's a simple scenario but most of the times I've seen like customers or some of our customers run into performance issues so the common scenario is I want to do a word count for the last 30 minutes I've sent a set a batch interval of 1 second so what's the problem the problem is you have like Spock needs to go back to the last say 30 into 60 around 1,800 batches to do the word count so it's a Neha and SPARC needs to finish all this doing word count across this 1,800 batches in less than a second for the next batch to kick in so this is a big problem so one optimization is choose your batch interval wisely like does it really have to be one second because if there are other operations that require it one second you can keep it at one second or you can bump up the batch interval to a like one minute or so to kind of make sure that spark is not processing too many batches because each batch is basically an RDD and that has partition at least one partition in it so it transforms to 1800 partitions the second optimization is basically doing incremental aggregations so here what you can do is as new batches come in you can just add them to the aggregation and the old batches that are going out of the window you can remove from the aggregation so so this is possible only if you if your aggregation has an inverse reduced function basically for a word count it is pretty straightforward so for the reduced by key and window you can specify the aggregation logic which is the sum followed by the inverse reduced which is just removing the counts so any aggregation function that has like a equivalent a reduced operation a sorry an inverse reduced operation you can do you can use incremental aggregations there and apart from the incremental aggregations the other common thing that developers must do is to use checkpointing of the data itself like you can do a word stream dot checkpoint and give a checkpoint interval so what so how this will help us basically spark will not go all the way till the last thirtieth minute instead if the checkpoint interval is 10 minutes so it will go only till the last 10 minute and it will read from the checkpoint and then continue its operations so in that way it's kind of the performance would be much better but the biggest drawback to that as you're checkpointing the entire data stream which can also be expensive so if you give a checkpoint interval of like 5 seconds that means like pretty you're kind of storing all the data in an intermediate storage for checkpointing which can be which can become a bottleneck so this is required only if you want to do deal with large window sizes and you can give a an optimum checkpoint interval which is at least I would say 5x the window size so the next common scenario is basically doing a word count doing a global word count or maintaining some state across time from the beginning of time basically like for example if the users are browsing through a website and you are collecting all the information and you want to build the user session like base you want to do socialization basically and kind of maintain a global state of all the sessions that are there so the biggest the two biggest api's that are available to do state management is basically update state by key and a map with state so map with state was introduced in spock 1.6 and that is kind of the preferred approach for doing state management update state by key still available but it's performance is basically proportional to the size of the state so for example like if you have a state of 1 million words and if you have a batch that is coming in which is like hundred words the performance of update state by key like is proportional to the size of the state which is like the 1 million words whereas map with state the performance is proportional to the size of the batch which is like the hundred words so there is an order of magnitude performance difference between like map with state and update state by key and map with state is a major thing is the main thing that people should be using for doing state management so I'll just cover the detail slight details about map with state so this is the example that I talked about where a stream of user actions are coming in you want to update the user sessions with state and then get these stream out of it and then keep processing it in our downstream so the processing can be just storing the state somewhere persisting the state and Cassandra or some other data sink or it can be further downstream for like again pumping it to Kafka to some other downstream process so the key features of map with state is you can provide an initial state so if you already have an initial but even before the spark streaming job starts if you have a initial state stored somewhere like in HDFS or Cassandra you can load them as an RDD and provide that as an initial state to map with state and then you can also specify the number of partitions if you have a good estimate of like the size of the piece key space like if you're gonna have one hundred thousand words or 1 million keys then you can specify the number of partitions to ensure that the state is kind of evenly distributed across the cluster and you can also specify a partition ur the default one is hash partitioner in case if that doesn't meet your requirement if you think there will be data skewness because of the way the key space is kind of distributed you can have your own custom partitioner to make sure that the key space is evenly distributed across the cluster and finally you can provide a timeout this timeout is useful like if you have keys that are not going to be updated in the last say and n minutes or n hours so like for example if you are building a session ization application you want you don't want to have sessions hanging around for more than like 30 minutes or one hour so you want to kind of remove all the old sessions that are not that doesn't have any recent activity so having a timeout will help to keep the state in a more pruned way rather than just keep it keeping it grow indefinitely over time so how that will look so you have a d stream and then you can say dot map with state is an operator on d stream and then you can specify you provider specification there so the state specification takes in the initial state the number tations partitioner and time or these are the full things that I talked about all of them are optional fields the must field is basically the function to update the state so what what is the logic to basically maintain the state and that is given by this function which is defined as follows so for for every event that is coming in the streaming job will call this function with these parameters the batch time for that particular event the key for the for the event in this case in the case of a word count it's just a word and the value would be 1 in this case and the previous state so the word counts so far for that particular key for this particular key and then the return value is basically the new key and the new word count for it so and one thing to notice this input key and the output key need not be the same so which means you can you have more flexibility in kind of managing your key space so so let's move on to the operationalization part so there are two the first thing is check finding there are two types of check pointing one is check pointing the data the other is check pointing the metadata so parts of it check pointing the data like talked about during the reduced by key and window this is required when you're doing computation across law a large window size so for example here there is a operation that that it relies on the previous states here but you can see here we have check pointed this particular word stream so it doesn't go back to all the previous stages instead it reads from the check point and this gray means these stages are skipped by spark so there is no computation there is no action computation that happened so spark just directly right from the checkpoint and proceeded so this this is one case where it will be very helpful to kind of have saved some perform I mean have some performance optimization if if there is a large window operation that needs to be done on ready stream the other part is checkpointing metadata so what kind of metadata so spark maintains all the current batches that are being run and all the batches that are queued up to be run so in this is checkpointing the metadata is required if you want fault tolerance and if you want a recovery from like from failures automatically like for example if the driver process crashes then if there is no checkpoint of metadata then Spock cannot recover from where it left off so you have to kind of manually or there will be some data loss that will happen so the best way to enable the checkpoint is basically say streaming context or checkpoint and then provide a directory to it so automatically all the active batches that are being processed gets checked pointed to the directory and whenever the streaming job crashes you can again while creating the context you can say get active or create and then provide the checkpoint directory so you can use HDFS and also s3 yeah so that's about checkpointing the metadata so the next is achieving good throughput so lot of times we've seen like oh I have like a 20 node cluster but the SPARC stripping job is still slow or doesn't kind of scale up horizontally well with the cluster my cluster one of the most common things is this simple streaming job which has like a receiver then a followed by a map filter and then save the problem here is there is one receiver that is running on one of the executors which is receiving all the data and the data resides in that executor and there is a map and filter operation now SPARC is going to use data locality to kind of schedule the tasks since the data is just in one executor so it's going to schedule everything in there as much as possible so there is even if you scale up the cluster the SPARC I mean the streaming job is not going to scale horizontally with the cluster one common simple solution is immediately after the socket stream you can do a repartition and then kind of distribute the data across the cluster and then follow other operations on it like map or filter or reduce stuff the one thing that with that as repartition causes a shuffle so there is still some performance overhead a more neater way of handling this would be to basically increase the number of receivers and then Union them so for example instead of having one receiver you can have five receivers and so that the five receivers run on five executors in parallel II receiving the data and then the further operations on them kind of there is more distribution across the cluster so this is an example for that like what where you create five input streams from the socket and then you can do a extract union of all those streams a union is a very cheap operation it's just an update of the metadata for rdd's to just say oh the partitions are located in these executors so you get a full stream of all the from all the receivers and then you can further continue with the further operations on the B stream so in the case of direct receivers so with Kafka and other similar data sources Spock streaming doesn't use the concept of a receiver simply because Kafka itself is a good buffering system so why kind of receive the data and buffer that in spark streaming so spark relies on Kafka to do the buffering and it just notes down the partitions and the offsets that it gets from Kafka to kind of receive the data so in that case the best way to kind of increase the parallelism is increase the Kafka partitions so each Kafka partition gets mapped to a spot partition so having more Kafka partitions means more spark partitions which is more parallelism and spark the other so far we've we saw like the parallelism while receiving the data and doing the initial computation but after shuffle if you still want to ensure that your cluster is kind of being utilized maximum for most of the shuffle operations there is a parameter to specify the number of partitions so it's always good to kind of specify that to make sure that the data is kind of being evenly distributed otherwise there might be some default number of partitions that might be set like 200 and that that will be the one that is used for all the shuffles so you might want to have more control over how you kind of shuffle the data around so as far as debugging a streaming application in the spark UI there is a streaming tab which is a transient one this tab is available only if there is a streaming job that is running on the cluster so it has pretty wealth of information about all the state of the streaming job that is running for how long the streaming job has been running how many records have been processed how many batches have been completed and also the number of events that are received per second and also it if you can expand this to kind of look at the rate of events for each receivers so if you have like 10 receivers you can kind of look at the rate of events for each of them to see if there are any issues as far as the processing is concerned since a spark streaming kind of maintains of q for each dag and the batch interval you might want to ensure that the processing time for each batch is always less than or at least 80% of the batch interval that is provided so for example here the batch interval is two seconds the average processing time is 450 so which is 50 milliseconds so which is good if it is like two seconds or two 2.5 seconds then definitely the spark streaming job is going to crash at some point because it's going to keep accumulating more and more batches that are getting going to get queued and at some point it's going to die and you can see all the active batches here so here there is only one but normally if there is batches getting queued up you can see lot of batches present here so that's an indication that like okay there is something wrong with the streaming job and you also have a list of all the completed batches like the time with the timestamps and the number of events that particular batch processed along with the processing time for each of those batches so if you think one of the batches kind of went bad or like went crazy you can for example so this one like took two seconds this one took three seconds so this is an example where like our periodic intervals I'm doing a checkpoint so that's why these patches are kind of higher latency is because checkpoint has an overhead but they're going to catch up with other batches going to be much faster so overall the amortized processing time will be slow still better but there will be occasional spikes in your in the batch processing time if you are using a checkpoint so you can click on any of these batches like you can go to the batch details page which has the details about all your input information like you have all the offs here in this case the streaming job read from two partitions from Kafka and these are the offsets for them and if you have if you want to drill down further you can click on any of the jobs that are there so these are the jobs that are run for this batch so and you can finally finally drill down to the number of stages that were there so these are the two stages that was run for this batch and this stage had like one task and this had like two tasks so it's a very lightweight streaming job and if you're interested you can further drill down from the stages to the tasks so you can click on any of these stages to drill down into the tasks so the tasks will have the kind of the lowest granularity possible information where these tasks are run on which executors so this is a good place to look to make sure that the tasks are evenly distributed in the executor so this is a very so normally this is a most helpful page like in terms of trying to find out if if the cluster is being utilized properly so in terms of key benefits of spark streaming because of the way like Spock streaming could use the job so you have like the traditional systems where a distributed cluster is there and then if there is an uneven unevenly partitioned data that comes in one node can become a bottleneck so so with Spock streaming that's much efficient because for every batch interval like Spock serving schedules the tasks so if there is one node that is kind of has an uneven partition so the subsequent batches can still kind of use the available resources to keep moving fast and of course you also for the same reason with the way the architecture works you get fast failure and straggler recovery also for free because if any node goes down the next batch can always kind of use the other remaining nodes and kind of discard that node so that the recovery is much faster when compared to other systems traditional systems so the other key benefit for using stream SPARC is basically streaming on spark is you can combine it easily with batch so if I have a data set a table or a file sitting in HDFS or Cassandra I can always load them as an RDD and then I can join with my stream so here I'm loading this file into an RDD and then I have a Casca stream that is coming in and I can say kafka stream got transformed so what transform provides is the underneath our D D in the D stream so now I have the our D D and then I can join it with the static RTD that I got from some external source and then I can do other operations this is very useful if you have a stream of events coming in and if you want to just tag the events with some data from external source so I just want to join it with some external source and then tag it with further information so that I can process it to the for the downstream stream downstream consumers the other most common thing is doing predictions on real-time streaming so you can do you can learn a model offline you can use any of the standard ml algorithms that are available like any classification or regression or our simple k-means in this example and then you can apply the model on your stream and do a prediction on a more real-time stream and the final thing is combining streams with sequel so you can also kind of a lot of times I've seen like people use their batch data set they run they use data frames or use sequel to do the analysis and now they want to move it to the kind of on their streams so the easiest way to do is you have a stream you can say for each RTD or a transform which gives the underlying RTD and then you can create a data frame out of that RDD and copy the exact batch code that you did into this function and continue with the operation so it's very easy to kind of test your batch code in a streaming fashion and this is one of the actual main motivations for the next version of streaming inspark which is the structured streaming which is coming in Spock - oh it's basically to say like I want to just do a simple streaming job I don't want to reason about all this other things that are happening like these streams and stuff all I'm doing is reading from this file and reading from this source and writing it to this thing doing some sequel operation or some ETL operation in between instead of saying sequel context dot load I should just say sequel context that stream and everything will work as as it is so that's one of the biggest motivations for structured streaming which is coming in spark - oh yeah yes so there so this is going to remain as it is structured swimming is a whole new approach for streaming and we'll kind kind of kind of continue to invest time on that to make sure to see to keep improving that based on the lessons that is alone from these streams that's all I had questions so you want to drop the tasks or kill the task or the there are ways you can do you can combine so there is a streaming listener interface that you have you can implement yourselves that gets all the information about how whether a batch is completed and if you don't get information from that listener interface for a specific batch you can then force a force a particular batch to kind of throw an exception so that it proceeds to the next one so that it's a very happy way there is no straight API that is available but yeah you can do it if you are king uh-huh okay Oh Oh like the same value of us yeah so one way there are different ways to do it but one way would be so basically one way would be to say like you can do a broadcast hash join so you don't need to explicitly broadcast your small table you can do a simple join of a large table with a small table and then you can specify a broadcast hash joint parameter so automatically SPARC will figure out how the right table is small so let me do a broadcast hash joint okay the best in that case would be to use the transform function so whatever you write outside of any of the operations doesn't get picked up by Spock streaming right because the scheduler kind of picks up only the dag of operations and that's why the transform function was introduced so inside the transform you can have a non-volatile kind of RDD which refers to your small table and then you can broadcast that or non-volatile any variable to your data set and you can broadcast that inside the spa inside the transform to make sure that and you can also make it more sophisticated because the transform also gets batch time so you can do it once every 30 minutes if your data is changing only occasionally oh that depends on ya the how frequently your data the small data gets updated yeah I'm not sure on top of my mind like the best way to clean that clean the data I need to look into that you can use accumulators but accumulators are not fault tolerant I mean so if a task fails and gets rescheduled it will recount so if you want exact correct counting accumulators is not the best thing to do instead rather use the disclaimer or DD to do the actual counting