BDSBTB 2015: Steve Hillion and Chester Chen, Enterprise Big Data Analytics with Scala at Alpine
so once again I'm Stephen hilly and co-founder of Alpine and Chester is our director of platform engineering so he's built the analytics engine part of which we're going to describe today what we do is we have built this platform that mostly medium to large corporations use if they want to have the power of a system like spark to do complex machine learning but don't necessarily know where to get started we want to talk today about the challenges we've had in working in and around these large-scale parallel platforms like Hadoop and MPP databases and how we've leveraged Scala in particular and also spark as a mechanism for for doing that while hiding some of the complexity from our end users so we'll talk in some depth about a couple of instances of that although we've actually done that in many different many different forms and functions here's a quick sort of very high level architecture diagram essentially what we've built is a parallel analytics engine that interfaces with large scale databases with Hadoop MapReduce with spark with cloud platforms as well allows you to do large-scale machine learning without having to write a lot of code and essentially to be ignorant of what the underlying data platform is not willfully ignorant but not have to worry about the complexities of that platform around that core parallel analytics engine we've built a library of machine learning functions often contributing those back to libraries like ml lib offering often leveraging those and then wrapping that within a visual workflow interface that users of say pentaho for example or to SPSS minor would be somewhat familiar now that's introduced a lot of complexity we have this very easy looking interface as you'll see here and that hides a lot of the complexity so how do you take these complex algorithms and paralyze them across a broad variety of different data platforms that's complicated enough but how do you then tune those platforms how do you monitor the progress of your job how do you bad data how do you deal with reporting error messages and so on and so on all of those things need to be encapsulated this end this in this analytics engine and meet as simple as possible for the end user so some of the examples of the things where we've had to innovate and and and and leverage scholar in order to make things more straightforward in a way that chess is going to demonstrate is for example building a simple visual interface that masts that complexity providing access to a broad variety of different data platforms but essentially viewing all data is sort of equal whether it's a table setting in a database or a park a file sitting on HDFS another level of complexity is we want to give access to multiple languages even though we've built this visual interface we want to make sure that people can sort of break out of it and leverage something like our or or spark natively or hive or peg or MapReduce or whatever it might be so we wanted to to provide language support we then wanted to go beyond that and actually make it extensible so if you wanted to provide specific functions and have those available to all of the end users through a visual interface you could do that as well essentially a plug-in architecture for these visual operators of course the kernel of this is actually taking some of these algorithms whether that's something like a logistic regression or something as simple as a row filter or a histogram or a box plot want to make those run fully in parallel and have the user not have to worry about how to tune those algorithms to make them run fast and then be able to do that interactively and be able to visualize and model just as simply as if you were doing this in Excel and then also be able to monitor and manage the progress of those is actually one of the most challenging things that we have to do today and that's one of the things the Chester will drill into in a second so the idea I'm trying to get across is that while the outward interface is very simple to the user it's masking a lot of complexity and we've really of all the languages and technologies we use I think it's chiefly Scala which forms the backbone of all of the programming that we do and spark has sort of the flat flagship platform for us for deploying these large-scale machine learning jobs so Chester will go into some of the tech details about how we've we've done that so how do we leverage in Scala and spark so we have using scholar and especially archive for many different things so for example we're using spray in a cot for the recipe is we're using al Qaeda building our acha our servers to to to get around the problems of you know this copyright licensing issues so we also amusing agha for Asian architecture you'll see in a second and we're using of course using scholar with spark but in this talk and because it we only get a 20 minutes altogether so we want to focus in one particular part so how do we integrating the BS the op i integrated with a spark so in particular you know how do we build up the communication communication channels for spark so here's that briefly this is our agent architectures so because all client have to support you know all kind of Hadoop distributions from you know my part to Johan works too Cloudera in a different version of these distributions and each distributions may have different protocols you know protobuf could be different an API could change so essentially from the same user interface how do you access different type of data source without you know different deployments so we're using acha you know to sending messages from tom cat to the agent these agent each agent essentially our caste systems allow you to talk to a particular type of Hadoop distribution or different version of how to diffuse distributions so and also these agents also talk to relational database in the we talk we support green plant teradata GBCs in other in others you know flavors of stuff we talk we support so with that in mind and say okay now how do we integrate this part and so in particular we want to discuss two pieces one is a different type of a spark deployment mode and the other one is a how do we build this task communication channel so why we mention about the deployment mode so essentially as everybody knows a spark has a different type of for the deployment mode it's a standalone mode that there's a deploy in the yarn or using muscles even in your yarn there's a to type the yarn line and young you know young cluster so if you look at these pictures we talking about young Klein a young cluster young Klein is labeija Laden the spark drivers on the client side in the spark jobs in the young container and well young clusters you know all the bus driver and is for jobs in the container so here's comes the problems so if you're using young cluster this essentially turned into a batch jobs so we have to use a young young cluster because our application is used by many different clients and they wrote all kind of spark jobs and we have no control of so if at this time somebody decided to write it in a bad one for example allocate aggregates a lot in a spark driver you crash the GVMC essentially crash everybody else jobs as well so we need to use a young cluster mode but the young class i'm essentially as i said it's become eventually become a batch job there's no communications no reporting there's a note essentially just wait for even around 10 hours you have to wait for 10 hours to get it back so we want to change that we want to see how do I monitoring where the jobs going you know if job you know round five hours you see something wrong you want to stop it right so essentially we added three things three major changes to the young clients in order to do that so we have essentially have to fork the spark young clients a module so the first one we add the yarn application listener essentially allow us adding a new listeners allow the the young container with start stop and kate gave us call back the second part in this second party is on this market young client will add a stop you know Kayla application AP is which is a simply take the application ID and call kill you know killer application and the third one is we add a callback for resource in you know resource callback allows to estimate how much resources the yarn has so this is a the first part adding young application listener essentially gives us some something like this first he gave me a young progress indicator so you know he tells me you know he got 0% ten percent and gave me a application ID which had in turn can use first stop the jobs the third one give a yarn applicant tracking your eyes so so in the if you look at the spark source code and yarn Klein essentially what do we change is something like this there's a method called a monitoring application which is a while loop just waiting the spark job to finish in there once the state changes we essentially call the application listener to notify application progress applications failed you know any of these state changes so essentially you got notified once the state has changed in an hour on the application side on the edge insider essentially once we get this call back we're using acha send a message to our applications and which can lend them to log it or do something else so what this functionality enables us to do this first part so you notice when the young application starts before the spark job actually starts we already have a tracking URL there which can bring you to the cluster page and have a progress bar tells you to zero or ten percent and also because we have application ID now we can we can use the stop button essentially called the API to kill the job if you want to stop it so this is a really great gable early respects it early in a feedback for the users you know even before the spark job starts so the next party is okay so go to the actual spark is a meeting the spark jobs as everybody knows if you want to spend submit a spark job essentially you neither provides the drivers of memories number of executives and how much memory you need for each executor zand you know the end of course so the problem is you have to essentially estimate these what do you need beforehand you know if you giving too much and job got killed because you know basically you know Sparky essentially have this code saying that if job is over the maximum of the in a container plus overhead you know- up overhead essentially job get killed so so and also you don't know what you know the data could be very small or could very large it so so these are very hard to control how much memory you need so what we did is essentially we modified the the yarn clients ap is to essentially passing a function this function is that we come young resource capacity which contains the max the memories the containers has in the end a number of course on the container has this memory this function is called from the yarn before job starts so this way this provides us this first we allow to reset the memories if the user asking too much of our memories because we that's that this point already know how much max memory the container has and we know what's the overheads so if you see you know I asking too much we just reset it to the maximum so you can continue instead of job get killed and the second one is if you happen to have the file size you know you already know your file size now you combine the way the max memories you already get it from the callback you can do a better estimation dynamically program Lee to give you a better driver memories and all in eggs extra memories so but there is a reason the reason the with the recent is sparking is especially the sparkle after 1.30 one and they have the added sparked enemy dynamic allocations so this allows you to basically 10 amps mark allow you to dynamically increase or decreasing the number of executives but you have to specify / executor how much memory and of course you have but at least this solve one of the problems but we still have to estimate the driver memories and so you know we don't we don't like fork spark so we will gradually reduce our our fourth changes but but I will still have to somehow find a way of asking me estimating the driver memory as well as how about the the the yarn listener part so this is our first changes the second part of changes about the task communication has mentioned that is a bad job in the young classroom mode there's no communication back and how do you find the error reporting how do you find a live progress report so what we did is this when we submitting the spark jobs essentially we create a sparkling client listener on the Asian side on our client side so this one has a sparkle URI we basically sent along with all the other configurations we send us part this uri on the spark job site from spark contacts the sparkling contacts actually have this actor systems we can create actor out of it so we create a clear color really messenger actor which can do a handshake with our client client side the spark listener and once this handshake is done the communication is established and then we can communicate after that so inside the cluster when the job starts we created something called application contacts which resolve the client client side sparkless interactor will create a new actor and then we could a logger using this and then relay the message over on top of that also created a spark listener and we will see in a second to get the progress the loggers essentially sending different types of messages and then our prints tape and also redirect the logger so this way is on the client side you can see different logs actually even you write a lock on the print statement or on the spark job so this is what it looks like so on the spark contacts this a listener buzz you usually by default you have these a job progress listeners in event log listeners which goes to Sparky UI and his spark history log you know separately we could a new one this job progress real a listener which sending acha messages to our agent in our agent in turn send a message to our Tomcat which using WebSockets to directly upgrade update our spark you I our Alpine you I so so this is an aspartate on for example and this this is this listener detecting the stage already finish that the task that the spark chopped stage finish once jobs finished we using the logger send a UI message to our agent so as you can you can imagine that we actually have different type of messages from the actor you know from the jobs logger send to the agent side there is a blog type there's a you I type in there's application type so the UI type could have a progress bar could be a start message or end task message or simply display something so on our client each inside we have this listener once you got a lock message we you know directly send to log 14 which is a locking on our klein sites if this is a UI message and we see it's a progress bar with display a progress with the tracking URLs if it is a counter with this valley counter so there's all kind of things you can display and if this object application message we essentially update the status of the job and you know all the other things so so with this way you we essentially develop a very general communication channel from the industry on cluster deployment mode and allow us to actually see without you know going to spark and actually refreshing pages and you'll actually see a lot of things going on so the end result something like this so when we start a job the spark you I essentially the equivalent Sparky you I directly displayed on our UI without refreshing pages so if you go to Smart View I you keep the refreshing pigeon see where the job has been progress here because of using web socket all the things the live streaming directly to you so the user can directly see this so so that's kind of just a we can you because we only very short time so will you focus on various subset of what are we doing so I have a lot of other projects with this party we're working on peace you know there's a machine learning projects we're working on and there's also a long-running spark job or working on we working on visualization framework called chiasm to allow us to basically doing the visualization talk to spark and that requires a long-running spark jobs and we're using acha to run at it with you know we probably could not report that I mean maybe next time our next year's sometimes so with that I turned back to you okay just maybe a final thought is that as we've been emphasizing the purpose of this just a sort of summarize this and give an example of how we've been using it one client in particular that I think was struggling to do this is a financial services client though doing sort of credit reporting credit scoring essentially trying to build predictive models around very large quantities of transactional data and customer demographic data and in order to do that they were leveraging a traditional platform that took them about 72 hours to refresh their models every I have no idea actually maybe every month or every week something like that by leveraging Hadoop they managed to reduce that time to about fifty percent what we were able to do is go in and rebuild those same workflows essentially rebuilding them leveraging spark but again hiding all that complexity so in particular the specification of number of X executors for example and the amount of memory that chest was just talking about or specifying what particular operations would run was all essentially taking care of the engine so really the purpose of everything we've been talking about is to try and encapsulate in a single engine all of this complexity that you need for doing configuration of jobs and submission of jobs and error reporting as well as of course the parallelization of the algorithms themselves so certainly this has been useful I think for higher level users who don't necessarily want to be exposed to to a lot of this complexity even though most of the people in this room would probably love to be exposed to that complexity so if you do come and talk to us thanks so much awesome well thank you um maybe is there any questions out here that I can pass out the mic I'm just curious about you I what do you guys use for the UI Scala jes maybe or not yeah oh what's the plan yeah you're talking to precisely the wrong people at one point last year we we want to see whether we can use in leveraging Scala GS and but a week we get no didn't get there we did I invite you know how he to give our talks and internally and as well so we did a few heck thongs and try it with us ecology as so but didn't get into a production can you talk about data visualization how you're planning to do that where do you do the transforms what maybe you're the wrong people to ask about what you why you're going to use but well actually this is a project a visualizer team project it's called a chiasm he's a ds3 based open source for a park and you actually can go to recce go to the github to find it so the on top of you know the it's a visualization use the ds3 and there's configuration so we use you know you can take the configurations actually change the graph and you change the graphing change the configurations the stuff hood we talked about the leveraging the interactive talking to the naturally accessing the data that's the pieces that we decided to run because we have unlike the other everybody else in the room probably in there either run standalone spark mode or on the metals and or a young clients we have to run young classroom mode so that that check the challenges is you know start a young container take off you know few seconds and then spark starts so it's not a good for a way of doing the interactive visualization so what we we decided do essentially to a long-running smart jobs so keep the job start running and in sending essentially additional messages without you know restarting a container so and that work is in progress and we haven't really finished it that's why we couldn't present it here so that work combined with the the chiasm open-source chiasm so gave us a way of basically very quickly to you know ties and interactively you know see the data is quickly you know change different shape a different you know view differently rather than statically display the results just calculated so so that also leveraging our workflow behind the scenes is basically saying you can I can run a single spark sequel jobs or basically I can run whole workflow and then visualize the depends on what kind of a you know type of work you want to do so actually there's sort of two layers to the to the visualization engine if you like the the top layer as you would imagine is essentially a JavaScript or d3 based mechanism for doing rapid interactive visualization of data and that you can or if you do a search for chiasm online you'll you'll see the open source code around that and can monitor the progress of it and then there's a back end piece which is essentially sampling data out of the underlying data sources so that they can reach the browser and be fifty fitted into memory we have a meet up for this on the San Francisco big analytics September 17th at Alpin data labs so if we want to find out more you can join you can come so there's no more questions maybe you guys can just join me in thanking Stephen and Chester for their talk