BDSBTB 2015: Xiaojun Liu, Scalable Analytics of Machine Data
Recording: BDSBTB 2015: Xiaojun Liu, Scalable Analytics of Machine Data
thanks for the intro so here I'm going to mainly focus on sharing some of our experiences through two years working with spark analyzing machine data so as a quick outline I'll first provide some idea of kind of the nature of the data that we collect from our users environment and then go through more or less three kind of tips or lessons they have learned either through scaling our system for analytics or how we use spark its API to program our analytics functions so most of our target users they run a kind of a pre private cloud setting using one of the many virtualization technologies such as vmware vsphere or microsoft hyper-v so in this setting the hypervisor provides abstractions of compute network storage resources as kind of resource pools that are shared by many virtual machines and applications running on those virtual machines and in in in most of the cases these applications are not really developed as kind of a cloud native applications they still assume a stable runtime environment consisting of about guest operating systems and stable storage resources so it's kind of but to improve a resource utilization and sharing the hypervisor do provide capabilities such as live migration of virtual machines to balance the load for example across many physical servers in machine cluster and also in terms of moving the storage the data storage for a application across different storage arrays for example now because of the increased sharing of common resources when there is an issue such as well a app crashed or a working machine it's suffering suffering resource contention due to like I owe contention on the backend array the increased level of abstraction in the system stack do post a larger challenge in terms of both to administrators when they need to say provide provision new applications or virtual machines for for their users and also for us in terms of having more variables in our analysis to tackle the potential issues now our setup of the whole service roughly starts with while we provide our end users with a small virtual machine that serves as a data collection engine that uses our unused management server API to collect data we call generically these data sources they could be for example VMware vCenter management server or Microsoft System Center server and then all the data that's collected gets uploaded to our backend which we run using Amazon's Cloud Service and then we provide a common platform that crew that provides data persistence and access ap is and all the different analytics modules are built on top of this platform the kind of analysis we provide includes infrastructure planning health checks dynamic benchmarking for example for users to know like the level of system utilization in in 30 environment how does it compare to the rest of our users so in terms of the volume of data that we collect right now even with a still growing our user set without getting over 100 billion data points per day including both performance metrics configuration data plans events that captures changes in our users environment plus the platform that we are currently still keep building on top of spark we our idea is to provide not just visibility of all the data that we collect but also actionable insights that our user can use to optimize their environment now the suit for the rest of the talk will focus on the plan at the platform part of how we use spark and the associated technologies to create the data persistence and access and a scalable analytics platform now so if I may borrow the framing of a famous computer science textbook all the components that we have shown that I've shown you in the previous slides are really about how do we understand the structure of machine data that we collect and how do we interpret in terms of finding meaning of the data that we can provide as to our end users as insights to manage their operations so the kind of a very high-level conceptual data model that we arrived at it essentially we have a set of objects they can be were two machines physical servers data storage devices with various kinds of properties for example are they can have a name for a physical server they can have the model their CPU how many sockets how many course we have memory size etc and we have observations of these properties over time so if we go through the consider three outlined boxes from the top to bottom the top box are really about data that tells us well what resources are there you know were used as the environments how many physical servers there model what kind of cpu they have etcetera and in the middle box tells us about how these resources are being utilized in terms of their well CPU utilization percentage or groups or fall say our storage devices what's the average latency that a user's virtual machine is observing and then the bottom part tell us about what changes are happening in use of the environments these changes can be either initiated by say a system administrator that powers on a particular virtual machine or by some automated management processes in this case say we immerse distributed resource manager that may automatically move perform live migration of virtual machines across different host physical servers to balance other load so we started using spark to build our analytics platform almost two years ago it's a fast it's a fairly quick ramp up our team developed most of the back end components are using Scala and the play framework for all the web services component so we really be among all the team members at the time we have about a couple years of experience programming in Scala and also our initial usage of spark is fairly straightforward to me I started with a mental model of while using spark as scholars parallel collection on steroids instead running on a single machine but this market similar programming model but runs on a cluster that you can skew out easily now some of the initial analysis that we have done include well as vmware or microsoft releases their newark new hypervisor versions we want to we can look at the trends from our users data how quickly users are adopting each of the new releases and also we can look at in the storage devices that are present in our users environment what percentage is our storage area networks with this network file system etc now a issue a lesson that we that we have learned is for our particular workload we have to keep an eye out on the skill in partition size or the size in of the rdd elements so for example some of our map or flat map functions are the input the elements of the input our DD they are pretty much homogeneous for example the input element could be a task description of analyzing the performance data of a particular data source it's essentially consist the identity of the data source some kind of a good and let's say the starting time of a day but then once you are map it to make this element to all the let's say the all the available performance metrics from this user environment then depending on the size of a user's environment that can vary by like over 100 x between well small lab setups too well large-scale production environments so it may result in either different partitions or different elements in the rdb their size or the and the the amount of work for further map etc faces to where we buy orders magnitude so but in most cases actually sparks a task scheduler actually use roughly even load distribution among all the workers as long as you have a sufficient number of these individual tasks or elements to keep them busy for long for a fairly long time but we do need to use occasional repartition to resolve extreme skills that may cause some workers running out of memory you know were set up now for our analysis the largest volume of data that we collect our performance metrics that provides us information about latency of any resource requests like allocating a memory page performing some IO requests etc now with the kind of environments that we collect data from there are actually two factors that boosts the volume of data in this particular category for example with running let's say thousands of virtual machines on the physical server now previously let's say you have a hundred counters / physical server if you are running a dozen virtual machines that immediately gives you well a dozen more of x 100 performance counters / physical server that you can collect 1 similarly similarly for using multiple containers running in the same operating system another part is now with virtual machine virtualization technology the compute network storage resources presented to the virtual machines are and software-defined using a hot terminology here then that actually provides a lot more performance counters for visibility to this kind of technology stack for example in a recent release of we immerse hypervisor so a virtual machine there are over 60 memory-related concerts alone that we can collect from to analyze well the whole memory how the guest operating system are using the memory and how the hypervisor are adjusting say reclaiming I used memory from guest operating systems etc etc so to be able to handle such a volume of incoming data we actually evaluated lated quite a few well-known open-source time series database technologies including open PSDB that's our first kind of production use and then one of its derivatives kaerios DB and also a more recent one in flux TB now it turns out with the default setup even with some of the optimizations that that we applied on top of those given the volume of data that we need to handle it's really not economical in terms of well how many instances we will need to run your Amazon to provide enough throughput for these for the incoming volume of data so what we let's say as a quick sketch of an example let's say we want to use Amazon's DynamoDB service to be able to handle a hundred billion data points per day it's kind of roughly on the order of our current data incoming volume now from Amazon's recent pricing page on their u.s. East region to reserve write throughput it costs about point six five cents per hour for thin rights for a second it sounds reasonable but once you dig into it a hundred billion data points per day translates into over a million per second that translates into a per hour cost of over seven hundred dollars or over half a million dollars per month and this is even before we buy some read throughput to avoid that being a right only storage so if you so as we research various similar solutions of how other users are dealing with similar issues what we ended up with is well we do need to have some in-memory caching layer as the input as the data collected are coming in and basically batch the right to a motorized the cost of one right to the underlying key value store across many data points so here are some recent blog posts that from signal FX or my power of how they achieved a very high our throughput in terms of resisting time series data and for efficient access and in the end we arrived at very similar solutions now so besides the data persistence that we are covered earlier the way we did the way we organized our analysis API and how we build analytic functions is really following the philosophy of the lambda architecture we're at a very high level a query against a large data set is simply defined as a function over apply it over the data set now in this case what I'm illustrating here is let's say I want to provide our end users a kind of a baseline comparison capability the middle of the time line represents some change for example like hypervisor version upgrade etc then we want to compare the before and after over a certain time interval what's the system resource utilization levels then in this case as long as we define our function as being able to apply to a particular source of data over a arbitrary time range then make an easily apply the same function across to time ranges before and after to change and compare it to see whether there is any abnormal increase in resource utilization while running the same workload as far as applications are concerned so then your work is because of the diversity in the machine data we collect we do need to provide analytic programmers to be able to apply functions too many kinds of data in our case in this case illustrated the four categories of can more or less static kind of data that tells what resources are there the relationships between different objects which virtual machines runs on virtual runs on which physical servers and uses which story the race for their data our time series and events that captures changes in the environment so what we ended up with is if you go back all the data elements in these are dd's are keyed by a kind of a key structure that depends on the particular types of data in the end what we found out is well we are actually tackling one of the hard issues in computerized it's really naming the data elements that we have so in your case the data key consists about the identity of the object virtual machines physical server and then the kind of attribute different names of properties the associations from virtual machines to host to physical servers etc and then the time range of this particular data element now when they need to apply a function that actually gather is a subset of these are our DD elements what we came up with is actually there's a very natural naming of the particular function application in terms of well I'm applying a particular analysis function say on a virtual machine or a data storage device let's say this function name is whether it is whether there's any resource contention on this virtual machine and I want to analyze for the last week as the time range then we can use the different data dependencies for this function to map each data our DD from a kid by the name of the data to the function application key and then we can draw in the different data rd DS by this function application key and then the resulting value in the drawing will have all the relevant pieces of data that needs to feed into the particular function application so right now this still feels a little convoluted but they do give us a systematic way to structure our analytics workflow those in terms of batch analysis and streaming so in terms of our current production usage we have an hourly or daily batch drops in production things for almost a year and we are actively developing the spark streaming based pipeline which are testing your staging environment right now and hopefully putting production later in the month so yeah that's all we have oh yeah that's any questions and for feedback please reach out to extra cloth physics com awesome so if anyone has any questions I can walk around so for your time series database is Cassandra right now I think the scheme that we developed for storing the time series database using a key value like structure our schema can actually work against most generic key value store right now we are using each base in the back end very cool very impressive thank you okay thank you cool well awesome thanks you guys maybe we can join me in thanking show John