data.bythebay.io: Hiral Patel, A Real Time Analytics Framework
Recording: data.bythebay.io: Hiral Patel, A Real Time Analytics Framework
so today I'm going to tell you a little bit about a real-time analytics framework we built at Yahoo what does real time alright so I'm going to tell you about a real time analytics framework we built that Yahoo what do we really mean by real time in our context real time is an event that occurred less than a minute ago and being able to see it in an analytics dashboard for example if I were to go to a the yahoo homepage the fact that I went to the home page and viewed the page should be visible in a analytics dashboard and practice it's usually much much lower than a minute it's an on the order of like a few seconds my name is Hiro Patel and i'm going to i'm going to introduce myself real quick here i'm a senior principal architect where i worked at yahoo where i worked on real-time media analytics recording for search and native ads platform i've also worked on Big Data various companies using both open source as well as enterprise solutions so a lot of the things that we see with the real-time now an open source have probably been around in like the enterprise for a more than a decade so today I'm going to go through some of our use cases our goals for building a framework a few example architectures and some of the tools and technology we ended up using our use cases generally revolved around publishers and editors as a publisher I want to see what how my content is performing in real time I want to see what's driving views how is how are the house the content doing in terms of recirculation are people going on one news article in going to another one or are they going off site how's the how's the engagement on videos for example and essentially they want to capitalize on high-performing content editors on the other hand want to be able to promote content that is currently trending or content which may become viral over a period of time so there is data science component to this as well so why build a framework right so Yahoo's publishing products group which is the group i work for has many web properties each has its own product management and development teams this means developer efficiency becomes very important we want to decouple the team so they can all work towards their analytics goals for real time for real time analytics yet provide for a common framework this means that we can reuse all the optimizations monitoring the application metrics across all of those use cases Yahoo we have many different technologies that are used for analytics and serving content our technology choices are generally generally grounded in the and what we can reuse and what we may be able to bring in so in general we tend to reuse the same technology we already have instead of trying to bring in new open source projects unless we have a committer on hand or someone willing to take responsibility for for maintaining that open source tool let's take a look at a simplified view of an example architecture where we're producing minute aggregates so traditionally you know people would when they want to see real-time analytics they would just aggregate per minute or / 15 minutes per 30 minutes whatever they could get done in the batch fashion and load that into a database and that kind of pervaded into the real time space as well because you know we don't have any data store in general that can actually handle the amount of events and updates that need to happen in real time so what we have is a lambda style architecture which you know which is where Nathan Mars talked about having the the speed side and the and the bad side we would do something similar we'd have events coming in as a stream of events into a storm topology where we generate in-memory aggregates and push them into a database on the bad side we would have log files that your copy do we run MapReduce jobs produced the hourly aggregates and load them into a database an alternative to to this is an architecture where we have a data store which is capable of doing real-time aggregations of raw events we just produce the processed events and push them directly to the real-time store now there are a few examples of this notably druid pino or kudu in order to make it easy to onboard new use cases we built supporting tools for consistency we needed to we needed a way to centralize schema management on both the real-time and back sides we reused an internal CI CD framework for storm which allows us to easily onboard new topologies through configuration to multiple data centers and environments on the bad side we wanted to make sure that that data was being loaded consistently so we built a data loader which was resilient to failures we also needed to perform data lookups on yahoo specific metadata things like site property and etc so all the downstream use cases would have this data available instead of having to do the same kinds of lookups on their own for the data lookups we build a fault-tolerant service which serve the data sets to our storm topologies the topologies periodically would check to see a new version of the data set is available and if so they would grab it and refresh it and the same was true for the the service itself it would basically go to Hadoop and see if a new version of the data set is available and and served downloaded and serve that the data is it the data sets on the storm side were were locally backed by leveldb so most of us our data sets were less than you know two gigabytes three gigabytes at most and so serving those out through a rest service was pretty easy to do and all the updates on the storm side Werder done asynchronously in a separate thread so it didn't really halt the real time processing in any way so so now that we have the supporting tools we needed to make it easy to build topologies so we built common libraries on top of a configuration based apology builder this allowed us to build generic classes for common for data ingestion processing and loading and reuse them across all of our use cases this also allowed us to centralize common serializers and deserializer 'he's compression codecs and all the different lookups that were used for the data set service another important requirement which drove how we built this framework was performance we needed to process a certain rate of ingestion from our our internal fire fire hose which we call data highway and also the processed events which were to be produced by all the different use cases this meant we had to do capacity planning around io network network io disk i/o and cpu utilization let's go through a example let's say we have a fire hose producing events at a rate of six gigabits per second we would need to pre-process all of our events and perform our common look ups and push those events to kafka if our replication factor was 3 and we had a 15 note broker that amounts to roughly 150 3 megabytes per second disk i/o on each broker if we had 10 topologies reading our pre-processed events from Kafka our network I across the 15 brokers would be approximately approximately seventy-eight gigabits per second which is a lot of i/o right so what can we really do about achieving this level of disk and network i/o across calcio cluster one of the things we started with was to start benchmarking of course we needed to identify the performance characteristics based on various configurations of asian message size and compression codecs so we took a POC cluster and ran these benchmarks here's an example benchmark we did and as you can see the higher throughput was achieved when the when as we increase the message size and obviously at lower lower replication factors we used all the benchmarking tools that kafka comes with as a part of doing this this veg park what about the i/o in storm right so there's network i/o and storm as data is transferred between storm workers it is serialized on the source host and deserialized on the target host we use in order to reduce the network i/o by default storm uses Java serialization which is bulky so we ended up using custom creo serializers with configurable compression codecs to reduce the size of the data that was sent across the wire we experimented with various compression codecs and ended up using LZ for h HC based on our results and of course we want to filter out unwanted events as early as possible in order to increase our throughput on kafka based on the benchmark she saw we needed to increase our message size since we wanted to guarantee at least once semantics we use Kafka's synchronous rights so micro batching was needed to achieve the required throughput we made the back sides configurable so we could easily experiment and make adjustments we also added two levels of compression each batch itself was compressed each batch consisted of a list of events and those events themselves were compressed and we also had for each of those events we also had a meta event or our subset of event data so the meta basically contains maybe ten percent of the actual events data which could be used to quickly filter if we were interested in this event or not right so this saves us all the cost of deserialising the event decompressing and deserializing the event this in itself when you're like searching for a needle in a haystack speeds up your your real time processing quite a bit so micro batching roughly added a hundred millisecond additional latency given a bad size of ten percent of the event crates of our event traders 1000 we would say our batch size would be 100 right and at that we would we would add an additional hundred milliseconds which is nothing really we also gain an additional ten to fifteen percent of compression due to due to the batching even with the two levels of compression let's look at an example batching both that we have it's a subset of the actual class I just took some important aspects of it it's it's an example batching bolt we would use in storm it's responsible for processing the income tuples and passing valid batches to the underlying batch processor and the underlying batch processor is responsible for managing the lifecycle of the of the batch it invokes all the registered event processors for each event and accumulates the batch of events and flushes either flushes when either the size or time limits are reached it also provides mechanism to access the the compressed event through lazy decompression and DC relationship DC realization of the actual event so we won't deserialize and decompress and unless we actually need acne to access that event the event processor is pretty straightforward it's just responsible for its business logic all so most of the developers using the framework just wrote a class that extended the event processor and everything else was configuration all so the event processor is responsible for processing each event and producing one or more events as possible but you may have one event that has maybe all the section data of for a given page view and you're doing analytics at the section level so you'd have one event which would then has maybe ten sections so you would end up producing ten output events for one input event so you're doing a 10x increase in volume there so all the outputs events were also you know part of the packaging framework so which which essentially allowed us to scale the the processing of events even if we had an explosion of 10x the event processor also has also can maintain state within the back slice likes life cycle through the initialize and finalize method so initially if you didn't have a real time store and you wanted to push one minute aggregates you just implement the initial initialize and the finalized methods and you do your in-memory aggregates in the process the record method so I've shared with you our journey of building a real time analytics framework it wouldn't have been possible without all the engineers on our team and other teams who contributed to it and i hope this talk has sparked some new ideas which will help you in addressing some of your real time use cases questions is that is this so it's dudes it was a built as a part of the publishing products group which is responsible for most of the web properties male is separate search is separate but it's also used by our search native ads platform so lets you use both on the audience side as well as the ad side you're all yeah