scala neville li 3 23 16

so today I'm going to talk a little bit about shield which is a scholar idea for Google Cloud dataflow and are just flowing after a 12 hour flights and jet lag so and i'm going to do like live coding them or so what could possibly go wrong right so Who am I I'm Nevel I work has qualified been there for four and a half years I mostly work on today infra for machine money and analytics I work with the team that built discover weekly and related artists and previous recommendations among other things I walk a lot with the data big data frameworks like scouting spark storm and right now I'm working a lot on Google Cloud since probably the past year and a half so a little bit origin story at spotify we be using scouting for machine learning recommendations for a while since two thousand thirteen and so about almost three years now and a few people picking up spark for machinery as well ad-hoc mostly and exploratory research kind of use cases right now they are like 50 plus users writing Scala code most of them from data science background so very little actual computing or like a computer computer science programming experience and well actually we called a lot of down with Scala we have 400 plus unique jobs every day unique pipelines and probably thousands of them running daily so it's a lot of process processing with scholar and Lolly last year we started moving into google hot it's being secret for a while and we just announced it like a few days ago and sink last week that officially we're moving to google cloud right now there's actually a google gcp next 16 conference happening in San Francisco right now at pier 48 today is day one and tomorrow I'm gonna talk again at our conference about how we use data flow and scholar but that one's more like a higher business ish talk and this one will be a bit more technical and hope I can demo more of the actual code in this project anyway I studied is a hack project data flow scholar because we like the data flow model it's really nice I'm going to talk about the model a bit later but we also like Scala when I have a scholar API for it so what is the data flow google introduced this idea like two years ago from the mill-wheel paper which basically unifies patch and streaming processing if you're into like bit data frameworks flink also has the similar model that unifies the two it has a directly under which runs locally on your laptop so you can do like testing development very easily it also has a dataflow runner which runs on the Google cloud instances and it's fully managed which means you don't have to like set up your own cluster do a lot of scripting operational stuff you just basically run the main method of your job and it will send up some requests to the cloud and it sets up the cluster and it does auto scaling even like scare scaling up and down the cluster is needed so that's very nice also people have been building or the Reno's for it so right now they are like spark and the flink runners in work in progress and also it's being incubated into Apache or as is being incubated into a party as a beam so basically the goal is to have this unified SDK that supports all sorts of different execution engines so we build this hair cable project and what's a hardest problem in computer science naming things of course and because it's a scallop Roger scholar framework what do you call it that's the first version because he neither s seeing dataflow so where to put it it's better than state on floor I just kidding so we call it she'll actually a co-worker who the Italian came up with this name because it's Lottie and a quick look at classical Latin that's like the kind the Pope speaks means I know I understand pretty fancy it's a scholar framework name of a scholar framework that starts with s c and n 2 sio so totally buzzword compliant so what is she'll it is the API on top of Google Cloud dataflow and the core idea is very similar to spark oh and also I've had some ideas borrowed from Scotty talk about the two frameworks that we like very much there's open source on github and it's all measures and apache license too so locate other options like we had before what's the main differences why not just went scouting on GC of course coding is a very popular it's used by a lot of companies it's very much you're proven and runs on Hadoop which is also very stable the downside is you have to manage this Hadoop cluster export if I right now we have a 2,300 node cluster the largest in Europe and it goes down a lot very often it's always overloaded because of the amount of data we have and it takes a team of maybe 5 10 engineers to to just keep it up and a lot of like late-night calls with Hortonworks or whichever your your Hadoop lenders it doesn't handle multi-tenancy that well it could be overloaded during weekdays and you're very low you to the utilization over the weekend there's also no streaming mode in Scouting while trader has something but you can you can say that by Renaissance on top of Hadoop and storm so basically had to maintain two different clusters and it's just another API thats well probably less popular than Scott in itself so it's not something that we've investigate into well as what if I we do use done but we so far we've only been using the very basic java api so there's also Spock Spock runs nicely on GCE you can set up cross this pretty quickly and supports patch and streaming mode on top of that it also has interactive and a sequel query it has also has a lot of libraries for doing machinery in graph processing even support a lot of different languages well the downside is the people who've tried spark a scale there there are a lot of tuning parameters they're like hundreds of knobs to tune and right now like the Hadoop cluster we have we're running young and torrance back on ya is just very very hard to scale up because of the of the resource contention I mean it is easier to run it on cloud like a new sog Google Cloud like you can just spin up VMs and set up your cluster but then you have to do all the other lifecycle management like if you run one cloud super job you have to you know make sure they tear down or if something goes wrong they have to look into it a lot of operational costs so that's why we look into data flow and Scala data flow me the pros we already talked about it's a managed service there's no operational costs so when you run a job you can run it on your laptop say give me 50 machines and i will send a request to the clouds being up all the VMS and thus auto-scaling run in the hood it has a built-in integration with all the other products in Google ecosystem some of them were already like pretty proven like bakery it's the system that I inspired a lot of other sequel on Hadoop kind of systems and it's being Google spinning using it internally for like more than 10 years so it's very reliable so also like systems like big table and there's also a simple unified model for better and streaming which I'll talk a little bit about later and of course the dataflow SDK is in Java and it's a very simple pie basically you have a collection like a pea collection which is a parallel collection and have have one type of operation called to FN which is basically a flat map of course people from a scholar water i used to have bodies you know nice high level trigger transforms like you know future group by also of all sorts of things and that's why we build it and then you have all these reusable components transformations code FP style coder you can reuse they also nice scholar libraries for doing like machinery numerical computer like briefs which is kind of similar to numpy and also algae board a trader library for doing like a lot of the parallel processing approximation algorithms and stuff very nice so this is the typical work on example in she'll if you've done stop before you can see that the code is almost exactly the same even shield contacts I mean the variable at the value name is just as see what I call a copy paste able of course this is a google library so we have to show a page rank algorithm it's very simple you have a collection of source and destination links you initialize page ramp to 1.0 and then over 10 iterations you read a shrill the rank and then do some dampen a dampening fracture and that's it in 13 lines of code and there's some fundamental differences so mostly in the in the data model for example in Spock there's our DD for patch and then the spark stream is actually a different slightly different model called D stream which is the sequence of micro batches or sequence of a dds so the API they're actually has two sets of API that a slightly different spark also has a memory caching mechanism data flow and Iran has a single p collection which does posts patch and streaming so impaired you basically have a collection of items that you process you know all honesty and the in streaming mode the this the items in the collection actually windowed into based on time stamps so operations like group by and reduce or whatever operate on a window by window basis and you have the same API on both Spanish and streaming which is really nice so you can write your job and then test it on a log log file using the time stamp from the log lines as a is a window mechanism and then you can just deploy the job with the data pub/sub input which is a real time streaming input then you have your streaming job so that's a very nice feature for data flow and next execution that's also very different there's also very different it did a big difference between the two in spark there's always a master node and a bunch of workers so basically the master or the driver drives the operation on the workers that's how you get this dynamic execution and also inspect for example there are four concepts called transformation and actions transformation basically happens only on the workers each worker process on its own or you know transfer data doing shuffling and one not a bit among themselves whereas action will actually take data from the workers and and back to it to the master and button the master have the ability to decide what to do next in data flow and edge scouting to some extent there is no master node so once you write your job basically a determines static execution graph and it submits to the to the cloud and then you can just shut down your laptop there's no driver whatsoever it runs on itself on a downside there is there's no no action basically you have to and there's no dynamic execution basically you have to to determine or two to design what you want to do in the job ahead of time so that's a slight difference between the two so what do we get from the sky idea Geo a PIAA of course me you have Scala and i also added features from LG board like you can do some an aggregate if you you a familiar with those concepts we get those are very popular here for scouting people and also added typesafe bakery with scala macros which i'm going to demo in a bit and interactive repo which also gone a demo and then last every time you save something into into desk or data store whatever storage engine initio basically returns the future of a chap something that you can reopen in the future so that basically replaces some of the missing features like being data flow like the fact that it's it's static execution and there's no internet more this make up for that so the last two yeah demo time let's see Carlos everyone see that is a big enough okay so so I built this a repo gr here so I'm just going to Java run it have some parameters basically specifying the project the google project and the data location in this case it has to be in European Union because we are Spotify European company and this is real data not a sample data all right now you have this interactive ripple kind of looks like Spock yes sort of poured a lot of the ideas mmm all right so what do we do a typical would count I was a cheat sheet so I'm just going to copy everything from here okay so if i do this is the better yeah okay so what I'm doing here SC as the context i get for free when i started repo text file like teen stock flat map you have a text file into a collection of lines to a split by whatever now alphabet you know character convert to lowercase filter out the empty ones to come come by value now you have a tuple to of string and lon materialized basically saved it to some temporary location that you don't need to care about but you can read it back and it returns the future so that's the f1 it returns the future of a tab of the type of my computation screen along nothing happens here because i haven't submitted my job yet and i'm using a local runner by default so if i do a close yeah that will actually kick off the job in local computation direct pipeline runner which means that runs locally now i have a result press one is complete because its local so it completes right away what do I have with the future what can I do with it of course you can wait for results rs3 it's the chap tab basically encapsulates something i have on disk or in some storage bakery or whatever I can't open later so I can just do value which gives me a decorator take I don't know 10 for each train turn that's that's your result what else can you do with the tab you can open an inner in a second pipeline so I have this major key word for new shield which gives me a new context a new watch new copy of the SC so remember that every SC a contact correlates to a single unique job so you get a new one for the new pipeline the pro or the plus side is you can have multiple jobs running a sink in parallel and you can join the futures things like that so now i'm gonna do i'm just going to copy the code in some terabyte IP so the first thing i did rip the basic work on the foot of the second one I'm gonna do actually I'm going to assign this rs3 to t which represents my tap my pace call here so basically I'm saying given this tab I want to open it in a new context filter by counts greater than 5 materialize do some other temporary location okay and sedo close now my f2 is complete wait for results you know the drill so value for take five maybe for each friend here I have it so I mean this could technically run in in a cloud as well which I'm gonna show in a bit but I'm gonna show the most exciting feature or it is the one that I most found off so here's your typical bakery I have a decibel query client beauty is eq so you can like read and write bakery tables so let's say I want to get rose from this beaker table which is a real table by the way so we can go back to this table detail in bakery you I this is the track level metadata of tracking Spotify so you have all these content tags from music labels oconus that's a company that will cover it that does music intelligence so all these derived data hundreds of fields and this is moderate size table of four forty eight gig 64 million rows decent size so i can see get rose give me you know some tracks from a particular date it doesn't really load everything to memory it's a lazy iterator so i give me like it iterator over the rows of course you do next to fetch window because the way bakery works everything is JSON so a glorified map of string two objects everything is object so i can do rs9 dot let's see what i want to get mass metadata try to type it and of course i am terrible at typing so you get a now imaging in a production pipeline will run for two hours and you know now exception of course not what does it give me a object i have to cast into instance or something that i think it is so table row you know so and so forth basically a lot of trial error and get let's say I want to get track you know another object yeah pretty boring right the next example I'm going to show is a big hurry macro which actually uses the scholar macro provider feature basically I can generate code at compile time so let's say this guy here from query and where's my query so I'm going to paste this decent-sized query here basically select a lot of columns which I don't know the type yet but barely knows because I mean the schema is in the bakery table definition so why do I have to specify the type again I shouldn't have to write so I call it class te track entity because i'm too lazy to type so it generates all these classes case classes which are totally typed safe so what can I do with this case class i can convert json into it right so instead of doing the previous version which rebel version i was doing this guy here this guy gives me Jason and I don't want Jason I want case classes so I can do BQ dog it typed rose te do I have to enter the query again no because the curry is already in the type it's a in a companion object performs a bit career in the background now gives me a iterator this crew is pretty pretty fast or actually it's already already cached so it catches the temporary result in a temporary table so I get it yeah okay I get ride right away now i do what do i do next we have te equals to this now i can do all these things like mrsa metadata ok it's option i know it has value so I just gonna I'm just gonna get to the cage you know what do I want to get next track maybe it's another option because it could be not able so get ya what they are seeing track ok track name now I have a track name totally type safe and what else can you do with this with all these options and stuff of course you can do for comprehension so I'm saying I want to get much metadata and then get track and a new track name of course I want to get audio attributes and from audio attribute so I don't get a temple now I do a yield I have an option of track and temple what else can I do is this I can put it in a real job so go back to this guy here so remember so if I've been running jobs locally in a local laptop but this case I want to run it for real do this no so so I'm going to use this new major workers called few options or ops so basically saying instead of running the default empty options which implies local ruin with this option say running inside Google cloud project of this project ID using this google cloud bucket for staging data and also using a data flow runner which is basically running for real in the cloud now I can get a new context which maps to this new cloud configuration that can run you know stuff for real and then I'm gonna copy that guy over so type a query gives me a s collection of this case class and then remember this guy for comprehension here so you can fall into flat map into option so basically filter out the empty ones or whoever that which you have a track that's missing you know track name or any of the fields I'm accessing so like this right and then let's say I want to track of certain temple so underscore to greater than let's say 100 sounds good yeah they're materialized to somewhere in the cloud I don't care hope it works it works ok now it's time to find out if it really works so when i do the clothes it's going to submit a job to the cloud ok let's see it's uploading three files to the cloud which including one for the repo one for the temporary code I enters inside the console and some other file that I don't remember what it is but that's the idea upload some actual jar files to the cloud and cloud google cloud VMs will pick up those jaws and execute them so it gives me a link to the cloud dataflow where I can visualize the flow so I even included the stack trace of you know where your so if you're running from a scholar for this will show the filename and the line number but in this case is just the console and whatever magic ID the console uses internally I'm not really you know expert in the compiler basically it's duct taping everything into your works but it sort of works pretty well so I have a job it's going to take like many or two to spin up the VMS and then it's going to give me some result meanwhile we can go back to the slides so since learned from building framework I mean I like how studying how these things work like scouting spark but it's actually the first time you know I actually try to build something from scratch myself and they're not you know to be learned from such experience serialization it's a huge pain yet it's yeah it's everywhere hits you yeah so what scouting spark storm or these other frameworks does is they use cry or chill which uses a lot of like low-level bytecode technique to serialize scenes and a reflection which works kind of okay in most cases until you hit you know we had objects that are not serializable then you have to duct-tape data flow and crunch which is another Java framework written by X Coogler based on the flume Java paper requires explicit coder so basically every transformation every time you do don knapp door flam apt or whatever you have to specify coder which is pretty painful for the developer and a lot of cases you don't need those coders imagine if you do two maps in a in a row it's going to happen in local machine there's no need to sterilize data but you still have to specify the serializer data flow does some JVM reflection magic to to sort of figure out the the coder but it doesn't always work which is another pain so in Scala we can cheat by having class tags everywhere it's pretty but you can do that sort of reflection and get you know whatever that's at senior ascription or in your pipeline and then apply the proper code or or sterilizer and of course you can always fall back to car which will try its best to to serialize things except when you hit you know serialization libraries like everywhere protobuf which does all sorts of great stuff that got it just not very civilized abou but that's a whole nother topic close your civilization it's even worse whenever you inside your lambda you reference something outside like you know from from the closure it's gonna screw things up at posing everything from you know your job class or wherever we use the chills closure cleaner but it doesn't work one hundred percent the edge cases where either fails to clean up certain fields or clean up too much and you end up with mouse inside your lambda me again it's mostly software engineering right duct taping you know making it in serializable transients you know until it works but still some edge cases report read was reading fun it took us a bit actually it was written by a co-worker who happened to you who used to be a hadoop ops but now that we don't have to maintain a Hadoop cluster anymore he's freed up and to you know write actual code another night scene for for building scenes on a cloud anyway so whenever I type in the in the repo right has to be serialized and that of course the scholar can so does not generate class files so again heck around so that we hijack the console class loader and then just dump everything in the console current session in into into some Java and a shape here to the to the cloud the macro type provider it's really fun I mean you're generating code right basically you have your types tied to the sequel code and in the same place you don't have to manage and separate repo or or separate build system for like compiling ever protobuf files and because everything is type safe if you handle it's like in the same way just you know handle options we scare me if I can pause it it usually works usually works pretty well it's nice also discovered intelligible it doesn't doesn't understand the generated code from Marco so you have all these red text IntelliJ because it doesn't realize the fields are being generated well that's why we have the repo if you if it worked in the repo it works you know in real code right so maybe not a huge deal so what's the status right now a few teams in spotify are using it already i mean we really like the fact that integrates with all the other Google ecosystem like you can read from decorate down to the table some teams already using big table for serving production traffic so it's nice where you can do all these e TR stuff or whatever data processing without you know going through your current Hadoop cluster because our cluster is pretty overloaded also want to include a peer and the repo usability I mean there are some parts of the API that are not perfect I have haven't finalized so could use some improvements of course you always use more clothing more tests and dogs that's another work in progress anyway it's open source so feel free to try it out and submit pool crests and issues actually let's go back to the disc guy here that's a repo so it's very simple the urs very simple spotify / chill it's also very easy to remember SC though I ode to my name is not register below either maybe they don't allow like two character to main aim but anyway if you go to releases page you can actually download the repo jar which is a self-contained ueber jar with everything so you can just java-jar it and you don't need a Google account to to run local mode if you run a try bakery I think we actually provide some public data sets that are free to query so you can set it up and just play around those features back to the job here it's actually look at a graph oh it's done so this job let's look at this summary took three minutes not too bad and at each transformation you can see the some basic statistics like number of elements processed and also a total size so pretty nice for troubleshooting I know the data flow guys are adding some more features for like tracking performance vm stats and whatnot anyway so now i'm back in the console ok I didn't have to restart the console even once that's nice so the night scene here is like I run a job with 20 workers for example and I realized okay it's not enough right you just change the configuration and run another job with a hundred workers I mean it would be hard to do in spark because you have to tear down and set up another cluster and will be like close to impossible in Hadoop because me who's going to do that anyway let's see 21 is complete please complete so where's my result that's this guy the future of my result wait for results that's just interested class with this fancy fancy meta basically does the future wait whatever k value maybe not for you let's take ten of them for each you get some results pretty nice if I have another minute I'm going to show some code window some work in progress code this is a job to walking progress but basically it does recommendation for running tracks so we have this feature in Spotify where you can open Spotify and run to the beat basically what detect the beach the tempo you're running with and recommend track so you might like that you know fall into that beat Tampa bucket it's a simple idea but what basically what we do is have a gigantic be a query that selects tracks with certain criteria you know energy level acoustic nice you know whatever criteria duration sort of things and for each track we have a latent vector from matrix factorization algorithm we use phul for visual recommendation and of course for each user we have the same vector so we can compare the compatibility or taste you know likely who the user likes to put a particular track what we do is you know do these two gigantic your queries basically selecting everything into memory and due to the rest hard to do stuff how to how to doing in Seco in the Scala code basically for the tracks we do a lot of filtering a lot of duct taping basic filtering by by heuristics or whatever criteria the data guys came up with and then group by by tracks or audits basically filter out the less popular ones now you have a candidate track a set of candidate track for each temple let me make it a little bigger and then for each user we also fetch the vector of that particular user so the next we compute a nearest-neighbor which you can do either brute force by basically doing a matrix multiplication you can do any fancy approximation algorithm we have a library called annoy which is approximately a tree-based or random forest paste look nearest neighbor lookup mechanism in this case I use a sh which is a relative locality sensitive hashing basically build the index for the entire Spotify catalog data suitable for running and then for each user you basically brute force on you know to the nearest neighbor look up and generate suitable tracks for running for each pocket which is a pretty huge data set cause imagine no maybe 30 40 million active users times 2010 pole pockets x 25 tracks per user and in the end of the day you dump everything into some gigantic storage like big table and then you can serve it you know tulip live traffic that's that's the idea now go back to the slides so that's the link just search for for gcp next 16 there will be more talks my colleague ego he would we share the same session tomorrow afternoon but he will talk more about the streaming part of data flow and pops up I would cover I basically cover whatever I talked about just now and that's it questions have time for questions I look see time for questions yeah true yeah yeah sure amazingly it's not so yeah so the question is if it's specifically for Google Cloud it is actually not when Google open source data flow SDK it runs our google cloud or local mode like the two modes they provide but then people build a spark and flink runners and we actually try to think Renee works sort of out of the box of course its work in progress and there's some tweaking needed but the idea is Google implement all these sources and sinks themselves which is independent of for if you if you look at a thing called spark they basically rely heavily on a Hadoop library basically to use Hadoop for reading and writing HDFS and whatnot the Google approaches that they abstract the weight of sources and sinks so all these speculative big table data stored or pops up sources in and out odds they working any runner it's basically a library that fetcher for example the bakery example actually just runs the bakery in the background using the standard Java API and then a dumpster query into some flat files and it reads the files back into into your workers so it can be technically any runner yeah yeah yeah it actually doesn't have any caching mechanism I see them as like solving different use cases spark is really good at doing interactive you know lookups or in exploratory use cases I mean the dance that is it has to fit in memory if it doesn't you run into problems and also losing a worker causes problem because you basically have to redo all the work I see Google build this scene to to basically solve the problem of building large-scale pipelines that does both batch and and streaming but not really for for solving like interactive query kind of use cases you know in mind so that's the major difference but I mean there are there are other ways of doing similar scenes on Google Cloud bakery is like if used wisely bakery can solve a lot of the problems I mean right now even before this a lot of our users are using the career with basically ipython notebook and if that works to certain extent about but they solve different problems yeah I mean of course Google started doing this like a few years ago I mean they had all these amazing infrastructure for a while certain things like beaker a big table that they'd be using house like for more than 10 years and those are like solid reliable and proven mean those work really well certain certain pieces they built from scratch for the cloud product for example data flow even though the idea came from google few years ago like whatever they use internally but data flow sdk is actually built from scratch so there are some rough edges I mean that's one thing you need to keep in mind but of course Google has like I don't know like how many engineers they have like fifteen thirty thousand so I mean they catch up really fast again and also that's a business decision right so moving to cloud means potentially you will be tied to a particular in this case you will be tied to a particular proud vendor any it depending on if it's something something you see as a risk in the long run yeah but of course if you want to replicate all these amazing features yourself it's not impossible it's just really hard like a curry you can probably have similar to a similar setup with in parallel presto on top of RK or whatever but I don't see them as nice as bikkuri not and it's not yet not yet you