SF Scala, Marek Kolodziej: Integrating Non-Reactive Legacy Code with Akka -- the Case of R
uh hi everyone um it's a real pleasure to be here and honestly i'm not quite sure what i'm doing after jonas's illustrious talks so i'll do my best to deliver on a more practical side of akka usage so the talk is called integrating non-reactive legacy code the case of r where r actually means the r programming language and i'm a machine learning engineer at alpine data labs here in the bay area in san francisco actually so we all know that reactive is about responsive scalable event driven and resilient at least those of you who read the reactive manifesto now uh just a quick recap of what that actually means because people used to say oh it's really marketing it's not marketing these are quite precisely defined terms so let's just do a quick review to see what we wish to obtain and what we have to compromise on in certain cases so event driven is essentially asynchronous non-blocking and it basically optimizes your application around amdahl's law because amdahl's law says that your application will be at least as slow as your part of the application that has to be serially executed the rest can be paralyzed and um if you have blocking and uh calls or synchronous calls you're wasting a lot of computing power so yes you cannot really distribute the um the serial part of your code but you can at least utilize your hardware better while otherwise your your code would be blocking scalable so as jonas said location transparency and yes of course you could scale it up scale it out by basically having a split brain between say java little concurrent on the one hand and say spring remoting or rmi or some other horrible thing in a distributed context but obviously that's not really very nice because then you have to figure out you know which part of the code uses one programming model which part of the code uses another one so it's not really very convenient um and uh you have to factor in the unreliability of the network so that's really the key and and and and you know we go back to these eight um eight fallacies and so on resilient so you have to isolate the failure and when you think about the bulkhead pattern or just bulkheads in an actual ship if part of the ship starts absorbing water that the whole ship shouldn't sink right you should degrade gracefully and um and the nice thing that akka provides here is uh not just the felt tolerance itself but the nice ability to separate the service and the failure handling mechanisms right so instead of putting your try-catch blocks everywhere you say here's my supervisor i i i put my failure handling in one place and the business logic is in another and that's really cool um so responsive uh latency is really important clearly as you're trying to scale that's that's a given but uh like you know mentioned that's not really a given in in all these impossibility theorems um and you have to gracefully handle congestion and that's sort of what reactive streams that that came in account 2.4 are about with back pressure and so on so this is what we really want now if everything were written in aka the world would have been phenomenal but unfortunately not everything is written in accounts so we have to deal with it so first of all not everything is an actor right if you're writing your own code you can say okay it's everything's going to be an actor but in reality you have third-party libraries and you're not going to write them and and sometimes even if if you have a lot of legacy code um you may wish to refactor it because you at least have control over it but it's still really hard and you know maybe you want to get permission to to do that refactoring so you have to basically just live with it now the the other problem is um there are lots of blocking calls in existing technologies whether it's you know jdbc driver calls to databases or you know calls to actually technologies that are not even on the jvm well obviously the database is not on the jvm either unless it's derby or something but but there are also other other runtimes that you may need to call to because they provide some functionality in this case it's r which i'm going to talk about in a moment then the other thing is the problem of long-running jobs so for example if you have a directed acyclic graph um and you basically have um your on a particular note in the graph and then there's a successor node and if and the node that you're currently executing has to complete to move to the next step well maybe the the successor um you know has a um that is waiting on this result has an ask call right which basically you get the future and you're blocking on on on that future because you don't really know what to do unless that particular piece of data arrives right so in that particular case yes you have a way to deal with with with waiting on a result but if your timeout is huge and your network gets partitioned before that or you have some other failure you don't really want to wait until the timeout um that would be really horrible so for example in the our case there could be jobs that are running monte carlo simulations or something and they're running for a whole day so you could set a timeout to one day but if your network gets partitioned or your vpn logs out or whatever in in five minutes then you definitely don't want to be blocking until your one day timeout runs out because that would be really horrible um so so so that's where i'm going to mention heartbeats because if you for example have your remote actor on death watch you can basically monitor for the terminated message and you can say okay well i have a timeout of one day but if the terminated message arrives that means that the phy accrual detector however it's configured because it's configurable um when it when it determines that uh that uh you know the other node is unreachable then it can basically uh time out you know basically end up the processing before the timeout and not all the failures happen within the jvm well that's kind of not our problem but i'm going to show you that in this particular case it's it's possible to uh to to restart these external processes and that's why aka's supervision is really great because once once um the the failure is detected i can actually use this java client for r to fork a new r process but in general it's uh there are lots of different problems that you have to deal with when you're integrating with other technologies i kind of feel like integrating aca with r and basically the whole jvm part of the application with r was a little bit like integrating maybe j2e back in the late 90s with system 360 or something i mean it's like that horrible but it had to be done because there was business value there so what is alpine i'm just going to briefly mention what is alpine because that sort of motivates this use case so alpine is an application that's uh that's web-based and you basically use widgets um to create your directed acyclic graph of of transformations and these transformations can be based on uh data coming from databases or or hadoop so in this particular workflow you have let's say a csv data source from hdfs called credit csv there's another one called demographic csv you perform a join you don't actually have to write a hive query or a big job or whatever you basically say okay this is the common column and i'm just going to do it you know an inner join or something and then you can do random sampling and and select samples and apply a decision tree so after you've done all your relational processing you can basically apply a machine learning algorithm such as a decision tree or or a linear regression you know apply after prediction you can apply your receiver operating characteristic or lift or goodness of fit or whatever so essentially it's a it's an end-to-end uh dag of of of of jobs from relational to machine learning and and you can do it all in this application by just moving widgets around and selecting options instead of actually writing code um so that's sort of what alpine does and and there's a lot of value in that and yes we have machine learning algorithms running in and and databases as stored procedures and we have spark jobs and so on but the problem is writing machine learning algorithms that are distributed is extremely hard it takes sometimes like three months to develop an algorithm like a random forest for example and so we have probably something like 30 algorithms but you will never replace r which has approximately 5 000 algorithms so um so uh the the goal here was to say hey we mostly do big data but for small data we want in an integration with a tool that uh that provides you with all these libraries out of the box so sometimes you may have small data but you you have the richness of algorithms so let's just uh let's address that use case as well um okay so what are the the cases for and against r because obviously it's a horrible technology to integrate with with the jv app so we have to uh uh you know uh i guess uh determine that that the the values for the the you know the arguments four are actually uh outweigh the ones against so r has five thousand uh statistical machine learning libraries and it provides the numeric gold standard for for lots of implementations because the people who uh tended to write to publish uh scientific journal articles actually um did their their first implementations in our so unless they had a bug you know that would be probably the default implementation for a lot of new algorithms and it has a huge community so you get peer reviews right a few people find bugs they're usually detected really early on and the other benefit is data scientists already know the language so so the other benefit of of the r operator was to essentially allow data scientists to do something that is not canned right because here you have a pretty ui and you can select options that that are provided but what if you want to do something arbitrary you want to run a bootstrap algorithm or something that's not built into the the uh these operators well if you want to do something arbitrary then you better provide the language that data scientists actually know so it would be either r or python actually has way fewer libraries and with scikit-learn and so on r has more more options so and and yeah in r you can really focus on the data science as opposed to encoding because if you wanted to run a neural network or an svm or whatever you basically just you know running running many algorithms is a one-liner once your data set is actually already there and you know writing something like that in java using you know even existing linear algebra libraries or something or optimization libraries would be really a lot of coding and uh so that was the value but the the the stuff against it is the runtime is extremely slow if you thought python and ruby were slow then then then try out r and then then you'll be happy about python and ruby and it's really memory hogging because basically everything is by copy there's there's nothing really by reference in r you could say that in haskell basically everything is by copy but the garbage collector is so fast that i think it garbage collects about one gigabyte per second so you really don't even think about it very much but in but ours garbage collector is extremely slow and it actually crashes sometimes so it's really not that great the runtime is single threaded so you could say you know oh we know lots of single threaded runtimes like python and ruby they're they're they're at least not concurrent they're actually multi-threaded but they have a global interpreter lock so they're still very slow but at least you can have you can have multiple threads even if one thread is executing at a time r is actually even worse because it's just one thread period um and and and you cannot even do time slicing so if you have a problem you won't even know that you're in an infinite loop because nothing is monitoring what's going on um so it's it's a lot of problems right and these libraries were written by not by cs graduates they were written by statisticians and because r was too slow people started writing them in c plus plus or fortran and you know you get seg faults because people you know overran the uh the index in an array or something so it's you have to really deal with a lot of nonsense but the thing is okay so how do we integrate this horrible thing that's act that has business value but it's just really not engineering quality into um you know an enterprise application essentially so so the and there are more challenges actually because rs gpl right so you cannot really ship it you know alpine ships its software it can also do software as a service technically but but it ships its software because lots of companies like healthcare and financial companies have have hadoop clusters that are locked up behind their firewall and and you basically have to ship the software you cannot ship the software if it's gpl uh the rserv connector which is the java library written by simon urbanek at atnt research this one is actually lgpl so it's you know considerably better but still there there were lots of concerns about licensing so how do you decouple rs as much as possible uh and then the distributed computing aspect so you really would want a cluster of our workers because you know if you have a concurrent database um then um the database queries can hap you know you can have multiple database queries happening at the same time if it's an mpp database you know hadoop if these are small map reduce jobs then you can probably have several of them happening at the same time especially if you actually divide the resources in your cluster to have concurrent jobs running and so on but you basically need several r workers and if you use this rserve api you're going to spin up several our processes on the single machine but you cannot really have these processes running on different machines you need some jvm to control these processes you can only have several processes per jvm but they're going to be on the same machine so you would really want to distribute these jvms to control these r processes and then the other problem is rest is really great for data but it's not really great for you know writing business logic i guess you know on the one hand distributed objects were a horrible idea but at the same time they had some structure if you have to basically figure out oh what is my end point gonna be and what are my four methods gonna do you know to basically abstract an arbitrary program that's really hard if you have aka and you have you know you can define special case classes for every type of message to do a particular thing that's much easier than thinking about oh what is my routing going to be for for this rest stuff and you know which of the four methods am i going to use and blah blah so and then the other issue is you cannot really request you cannot make too many requests at a time because r expects a lot of data to be coming into it you know the the the an arbitrary node on the cluster is not going to read you know a csv file from disk you have to send it the data so you either do it through rest or you actually have to push the data throughout as an actual object so in either case you want to have um either back pressure or you you want to have sessions and basically say hey this worker is unavailable until until it's done with its you know several hours of monte carlo simulations and uh back pressure actually would have been preferred but the problem is that we're using uh the akka version two two three which is the one that spark uses otherwise we would have a jar health problem so unless spark actually catches up with with aca and and uses a newer version that's a little bit of a challenge um you know you could start using custom class loaders or some other crazy thing but it would be nice to just have one aka version in the application um so back pressure was out of the question unfortunately for now um and uh fault tolerance okay so so r fails all the time right you have seg faults because the the statistician wrote porsche plus plus code or something you have network partitions so or or you could in theory your vpn could disconnect whatever so so you have all kinds of problems so what are the solutions well first of all from the licensing perspective because without that it wouldn't make even any sense to even try it out right if you can't ship it then you just don't do it so akka is apache 2.0 and our server is lgpl but people said well let's do a license with gpl what if the guy removes the the lgp the lgpl license or something so so essentially the best way to to decouple it would be to say okay i have my jvm that talks to r and i open source that part of the application because it's just a connector to our really that most of the business logic is on the albind side so we open source this connector that that that communicates between the uh the closed source application and and and this open source component that simply controls the r workers um so that can be open sourced um but then because you have loose coupling you can just send these open source messages between between the application if they're open source and they're you know let's say apache licensed or you know creative licensed or whatever you basically are you can reuse them in your in your closed source application so actually provides this decoupling for from a licensing perspective which is pretty cool um and from a distributed computing perspective aka provides location transparency so again you don't have to think about oh am i doing java it'll concurrent am i doing um uh you know some crazy rmi or some other you know spring remoting or rest or something like that right um and um the cluster api actually would have been really nice however again we're using an old version of akka so that's not really quite available at the moment and as i mentioned you can structure the the provide structure and semantics by having specific messages for specific problems and again um reactive streams or back pressure would have been nice but we have an old backup version so we have to deal with that ourselves um from a fault tolerance perspective the rserv java api actually allows you to start new r processes to basically fork them the moment you create an r connection object so that's nice and if aka handles fault tolerance then basically upon failure you can use you can use the the the error handling to to create new r processes the key thing i think that akka provides here as well is that firecrawl failure detector because if you can subscribe to terminated messages by by basically having death watch over the remote actor it's not just when that actor terminates in an unclean way it's also when the the network disconnects so so you will get locally the terminated message when when the network is no longer available and that's great because in many cases it's not about guaranteed delivery or anything like that you at least want to tell the user hey your network is down or maybe um in this particular case it's not a you know banking transaction or anything like that you at least but but you want to tell the user in the ui you actually need to rerun this workflow because a failure occurred so if you could do at least that that's that's where heartbeat actually helps and it also prevents you from waiting until the timeout is up right because if your timeout is one day and your network disconnects after five minutes you would rather actually terminate the flow immediately when the when a disconnection happens um and yes we don't need exactly one semantics and all of that stuff that's really kind of fictitious storm tried it and i'm not quite sure if it worked out okay so sessions okay this is a really horrible horrible horrible solution but uh basically because because there was no possibility of having um of having this back pressure and the and and the data sets are actually moving through aka they're not moving through rest because the rest api uh of the back end of the application wasn't complete because you know we weren't exposing it to external users so we didn't really pay attention to that sessions basically are sort of like a in this case like a poor man's version of um of back pressure basically um you know you get you get the consumer actually sends a message back saying oh this session is done send me new data but of course had they had the back pressure already been there um had we been able to use a newer version of akka that wouldn't have been necessary and also had rest been and used instead for data movement then then you wouldn't have to deal with that but that's coming in the next release of alpine's product also the important thing to remember is in akka you have to send set the message sizes so if if you have a configuration for two different nodes and you buy you accidentally set the two different message sizes then it may be the case that you know the message can get dropped because the the message size was inconsistent of course you know it's it's in your best interest to basically reuse the configuration to be consistent but i'm actually checking that the other server has the same message size if not i'm basically taking the minimum of the two it's really trivial i mean you just send a case class right you just have it have another case in your pattern matching to to uh to to pull right because you can get the value of the message size from the typesafe config so that's again kind of hacky but it's it actually helps you prevent really stupid mistakes on behalf of users who configure two nodes differently let's see so all right so let's move through that okay so this is really basically a very simple diagram the the alpine web app is running on one node um and it actually communicates with a so-called alpine agent through aca as well and the there is the reason for this agent architecture is that we have to support different hadoop versions map r and cloudera and pivotal and hortonworks and apache and the other problem is the problem is really not the vendors except for map r which actually uses a different file system but the real problem is that between different versions of um of hadoop the rpc protocol for hdfs changed so you really you you have the same package name right and you end up with a jar hell you cannot really load all of these packages into one one node so you essentially end up with um with with agents and and depending on the hadoop version for a particular workflow you communicate to a different agent and in fact you could have different clusters for the same user and one job can run on cloudera cdh4 another one can run on pivotal hd2 or whatever and and it can all work because of this agent architecture so the agent and the the web app actually communicate through um through aca already and then the agent talks through aka to the r server which is another jvm and um and that jvm um has these r connection objects which basically provide the tcp bridge between um r and and the jvm and and these are and these actors within the r server control the r processes so if the r process dies you know and and the the uh the actor that communicates directly with r doesn't actually handle the failure then its supervisor will and it will restart that worker and so once you get all that fault tolerance and because these uh these are connection objects actually when they're instantiated they fork a new um uh our process that provides you with a way of dealing with the problem and the process actually fails which is pretty often but because because you have data in in hdfs you know or in the database the data won't get lost right the only problem that you have is that you have to rerun that workflow but you're not losing any data and these are batch processes anyway so it's not a big deal the the big deal is really coordinating the whole thing and making sure that you know you don't have dangling resources and so on but so that's where akka really shines and and and this coordination really so you basically set the number of actors and the timeout limit and you know your remote deployment with protobuf sizes and so on now i guess it's too much code but essentially you have this rserv master and in this particular case it actually handles the session so it's kind of like a custom router um i had it had we had we you know the back pressure we really wouldn't have to do that um and other than that it's all really simple you have a supervisor and so on um and in this particular case i'm just showing you this this situation where you basically check whether all the sessions are busy or not and again that wouldn't have to be done if we had uh the akka version with with back pressure we you would you would just send as many requests to the to the queue as possible and it would they would just you know get seriously addressed um yeah yeah i'm basically almost done it's it's it's really essentially the end so so yeah this is basically a case where you just check the the the frame size for protobuf um on two nodes to make sure that you're actually not you know you don't have a bigger frame size on one node and then the other node will actually fail to receive it uh let's see and you have a supervisor which handles all these exceptions and this particular our actor actually processes these requests enriches the r script a little bit um and and is able to start the worker and and send the data to our sender request get the results back so i mean i'm not really going to go through the code the slides will be available uh let's see but the key thing that i wanted to also point out is that aka has a phenomenal test kit and it's really great not just for trivial cases where you basically have one request at a time but it's also which is for unit tests but it's also great for integration tests because you can basically test concurrent behavior and that's really really nice the one thing that i wanted to point out here is you could do something like like i did here which is you could actually mock let's see do i have the mock here yeah at the top i just used makito here because i i didn't want to get bothered at i didn't want the unit test to become integration test so i i basically mocked the behavior of the r process which is not really available at the moment on my on my dev machine and of course at integration test time you can you can make calls to the actual r process but you know makito and scalamock and all of these things play really well with scala test and and and scala test plays really well with um with uh the akka test kit and it's really important to test out these things because as you know you know we have reliability issues and concurrency issues and you just want to make sure that your application works really well and it would would have been extremely hard to test aka without the test kit and the test kit is really phenomenal so i just wanted to just say that that i'm really glad that it exists so i guess future improvements data movement through rest when it actually gets fixed in in the back end because the reason it didn't get fixed is because we're not really exposing it to anybody else but we will uh pretty soon and then a replacement of sessions with reactive streams and actually come on which is which i guess replaces the type safe console for monitoring actor systems and about a week ago um i got news from ivan topogniak who develops come on that it's now also supporting they have a beta version of of support for uh remote actors and for um and for cluster for the cluster api and that's really great because so far you could only monitor actors on one machine and uh so i guess i can i can just skip all this but um you know akka really made this so much more convenient thank you very much i just want to say one last thing actually too first of all alpine is hiring and we're looking for machine learning engineers who work in scala and java hopefully justin scala for the new code but we still have you know some legacy java code to deal with hopefully we'll be porting it over old scala data scientists who mainly work with are in python and front-end developers unfortunately it's mostly ruby on rails at this point some jsp on the on part of the application i wish it were play but but that's what it is um and also i wanted to make an announcement on behalf of michael slin who is the creator of scalacourses.com he is looking for reviewers i'm actually volunteering as a reviewer i mean the only thing you get is is you know gratitude and mention you get mentioned on the slides that are basically in the in the content of of of his um courses that that you are a reviewer you know there's no monetary compensation but i think it's really important to share the knowledge and and i'm basically just volunteering as a reviewer for him and he asked if i could share that he's actually looking for reviewers at this time so i guess that's it