scala.bythebay.io: Neville Li, Scio - A Scala API for Google Cloud Dataflow
Recording: scala.bythebay.io: Neville Li, Scio - A Scala API for Google Cloud Dataflow
all right since thanks for coming so today I'm going to talk about this framework or she'll it's a SCADA API for Google Cloud dataflow and Apache be how many here use any type of cloud AWS Google cloud a lot of us so yeah Oh make sense so let's talk a little bit about us I work for a company called Spotify we do music streaming we have a lot of features are the users seem to like music recognition features like discover weekly we actually also do a lot of trip data for music labels or artist managers for label and the user stream analytics and we have a lot of users about 100 million in a lot of markets and we have a fairly large Hadoop cluster a 2,500 node secrets as of today probably 2,700 of one of the largest in Europe in Europe and we process a lot of data so Who am I I being a qualified for little bit more than five years I started in the New York office when we launched in the US and I did a little bit Hadoop before that Yahoo search we did spell check and search related algorithms so as well if I mostly work on music recommendation and also data infrastructure actually I'm pretty new to Scala because what if I used to be a mostly Python shop and we picked up Scala in 2013 mostly for data science and that's when I started learning scholar and using it so a little bit origin story we used to do mostly Python we had this framework called Luigi which is a framework for post job scheduling and also Python streaming MapReduce and being around for a really long time and still there are a lot of companies using it but it turns out writing my producing Python is a pretty bad idea especially a few processing a lot of data a lot of structured data e-ever or whatever schema data type you have so around 2013 we picked up scotty and moved a lot of our pipeline over we also have people using spark mostly for machine learning use cases and also storm for real-time recommendation and at target II so those are the type of use cases we do port a tunnel it takes this qualify and we have a lot of employees using Scala about probably more than 100 130 developers writing data pipelines on a daily basis and we have a lot of jobs unique jobs so yeah earlier last year we made this decision to move everything to the cloud it's a big you know giant leap of faith at the point we had maybe six five six thousand machines physical bare metal hardware machines so it was getting to a point it's hard to scale and Google as we all know it's pretty good at you know scaling infrastructure and they have a lot of pretty solid data infrastructure products so we decide it's a good bet to move to Google and leverage there's technology so that comes goes to the question what else dataflow and being we have to look at the evolution of Apache beam from the very beginning around 2004 Google released asleep on MapReduce which inspired a lot of open source technologies like MapReduce became Hadoop there's also closest which inspired HDFS of course BigTable which inspired Cassandra and HBase and alone maybe two three years ago they publish a couple of more papers on flume which is a higher level data pipelining framework and also Newell which unifies stream and batch processing so the idea is that batch is really just a special case of streaming and you can do both of them in the same unified API the open sourced dataflow SDK and as part of the Google Cloud products and also later donated to Apache which became beam so what is beam exactly of course it introduces the unified streaming and batch programming model it also has this concept of SDKs which the developer used to write data pipelines right now there are two SDKs Python and Java the other languages the in purpose and then USD the pipeline that you write in a Java or Python SDK gets translated to one of the many runners supported right now there are three flink Apache Apache Frink Apache spark and Google Cloud dataflow there's also approach apex and gear pump be under review right now I think there are a couple of more runners being contributed to being as we speak there's obviously also a local runner for testing and also running jobs on a single laptop so beam also try to answer the or ask the four important question about data processing and streaming so what what is the computation you want to perform on your dataset flemeth map filter group by whatever where is your event generated in the timeline that's usually the timestamp of your log message obviously when in processing time are you computing these results in batch that doesn't really matter cause we just wait until everything is ready and then start processing in streaming it's more of an important question because we want to we usually want to process data as soon as they arrive but there's also always the possibility of losing data or dropping late data and how ask how do we deal with those cases so there's a simple way to visualize these four questions in batch you just process everything all at once in this case probably a sum of numbers in a window batch mode you package your data into obviously hourly or many windows and then process them one at a time but then there's the question of streaming in this case we have a red event there that's arriving late and in this case were dropping it so the event time and processing time it's not always in sync and the last question how do we deal those cases you can either accumulate update or maybe job late data so those are the four important questions about unified patterns Remy model so this is the Apache being vision obviously the users can write pipelines in any language they prefer which gets translated to any runner that's supported and then it should run on any of these platforms that's a ideal situation but we are not quite there yet and then I bet a lot of us here have used scallion or stock so it's probably easy to compare the data model of being was a spark so in fact we have this batch mode which and the abstract data type is a DDR resilient distributed data set and there's also a tea stream data type for a partner for spark streaming and really it's just a sequence of mini batches or mini rdd's so you know you end up with two sets of api's for patron streaming apart a party spark also has this explicit caching mechanism which is useful if you do a lot of iterative machine learning algorithms but could also potentially cause problems because you have to deal with memory like Java GC problems team has a much more simplified data model there's only one after datatype copy collection which handles both streaming and batch obviously they end up with one single API you can write at once and it should work in both modes elements in a pea collection have time stamps increase the timestamps and also window assignments so if you can't consume me from Kafka or pops up the producer might generate timestamp for these elements if you consume me from a text file log messages then you have to pass them yourself and assign the timestamp for the photo elements mmm another big difference is execution so inspark there's usually one master and a cup of executors the mass mass master will dynamically send messages to the executors through control playing materialized functions to perform transformations so like if you do a dd-don't map that's a transformation sometimes you want to get data back from your a DD and that's called action in spark so basically that triggers a computation and fetch data from the distributed memory caching or these workers back to the master data flow on the other hand has no master so you basically write your pipeline aesthetically generates tag graph for your computation and all the only thing you can do is transform then you submit your job which executes in a cloud you're completely detached from the cluster when said you can only get your data back once the once the computation is done so there's no concept of actions as you can see these two function signature compare value in spark is action it returns the e memory map count by value in Sheol on the other hand is transformation so it transforms one a scripture into another description so why do we use it I mean there must be some reason right so compared to the other options we have before data flow we actually we still do that we still use coding a lot in a company you can run around a Google cloud as well obviously the Vantage it's very mature and reliable there's a lot of company using it it runs on Hadoop which is also proven to dance I obviously you have to manage a Hadoop cluster and it's not an easy job managing like 2,000 plus node cluster you have the problem with the multi-tenancy of resource sharing a utilization there's no streaming audience coding obviously is some input which runs post on Scotty and stone but that adds more complexity right you have to maintain two separate clusters there's also no integration with the rest of GCP ecosystem like bakery big table or use google cloud products that we want to move over to spark has a similar situation spark has more features and supporting bad streaming Seco ml a lot of different languages has web you guys like sapling and spark notebook the downside again its cluster management or operations we have a lot of data scientists one not engineers and it's very hard to expect them to speed up and hear long clusters using a lot of scripts and then do a lot of tunings by hand so that's a problem especially if you run a large ml job that has a lot of in memory caching that's where a lot of duties reliability problems pop up you have workers dying due to GC timeouts and failures and triggers recompute each computation of the entire tag so those are problems we've seen and the last problem is also ecosystem there's no integration with the Google products so it's hard to to to integrate so that's why we picked dataflow and then use it with Scala obviously dataflow comes with the Java API it's a fully hosted and managed service which means you're right you're coding your local IntelliJ or Scala repo and you submit a job it spins up a cluster in a cloud automatically fully managed and handles like things like auto scaling shutting and then it comes with adaptors for all the Google cloud infrastructure products also another selling point is the unified patch and streaming model and by the way for those of you who try to think the flink streaming badge unify model is also inspired by the same male will book that dataflow is based on obviously you're here so you like Scala obviously we want to write Scala instead of Java functional programming is a good fit for data processing and we use a lot of Scala libraries for numeron call statistical computations law likes briefs which is like numpy for Scala and also LG but for a lot of the parallel approximation computation so this is what it looks like basically we took the Java SDK or the goodness from the Google cloud and put a bunch of Scala Apeiron wrappers and utilities on top and to make our life easier so the name shield it's pronounced shield that's ecclesiastical our team which is the type of the Pope speaks that's what I I'm told so pretty classy it's a library for for Scala data framework that starts with SC and ends with IO and buzzword compliant unfortunately I couldn't register the SC though I oughta maintain him for whatever reason anyway it's public on github and the Apache License 2.0 so check it out a couple companies using it already and so far they seem to like it this is what what count looks like in shield the API is heavily inspired by both Scotty and Spock basically we took the bits that we most of our users like from the two different frameworks and put them together and it's probably 80% copy-paste of all from stock so another bonus point it's a very easy transition for all of our users and of course page rank 10 iterations of page rank you have an input collection of tuple to source and destination strings you map out the links and initialize the rank at one point zero then oh well tear trations redistribute the rank and then you apply a damping factor of zero eight zero point eight five that's page rank in a couple of lines of code so that's what Scala people like easy to write preference easy to reason about so why do we like it there are a couple of features that our users seem to like a lot types a hickory dickory is the Google Seco big data analytics engine which can process like terabytes of data in a matter of seconds beakers actually based on the Google internal technology called Dremel which is also the same paper that inspired Apache packets or columnar storage column projection predicates push down those kind of ideas they all came from the dremel paper the Google bigquery API gives you rows in a form of table row a typo table which is really just a JSON type some map glorified map of string to objects so it's hard to work with in the in this case we just build some macros to generate because you're performing a correct so you already know what the return type results schemas so we just generated case classes and combo to us from the query at compile time and you get his classes which is type safe and it just removes a lot of the boilerplate now point exceptions and just guessing and redrawing those kind of paint when dealing with run type data we also build a repo which is again inspired by Spock you can interactively pull data from the Google cloud from cloud storage the table or picker e-reading real time gonna give it a demo in a couple minutes but it's just a single jar there's no it no installation you download it you run it and use your default credential again because as I said earlier there's no class to manage money invade of so you don't have to spin up a cluster and you know run the ripple or drive over there you just run a job locally and then and talks to the Google cloud service and in spark if you save something to disk the the save method save as text file or whatever the original unit because that's the end of your pipeline you're done you're saving data to disk and there's no more processing outwards but initio we added it's a future based abstraction so they actually return future which will be completed once your data pipeline is executed and completes and then we can use that to to orchestrate like multiple pipelines it's this example you have a future from a first pipeline job close it then you reopen the written future in a second pipeline there should be a cache so there's a common Hadoop pattern and we just really implement it in data flow because we have access to the to the cloud storage or HDFS we can just go fast to workers and load them into memory in whichever form we like and we do that quite often with like binary tree lock-up trees or hash table or like look-up tables and a couple of other goodies obviously data flow has a web UI and then we mapped the Scala function names source code file and line number to the web UI so it's easy to visualize we added a bunch of goodies for mercury and HDFS sauce and C which means you can really update a flow job in a cloud also read and write your own premise or your your loop trust the inner cloud and enhancement for like metrics and BigTable so these are the other stuff we added on top of the Java SDK alright so let's try the demo you guys you hear me I have to turn off the mic anyway let's try this so I just installed a job your home brew it's already there you run a command is the text big enough yes all right so I have a console it runs on my lab a local laptop they picked up my Google Cloud credentials also comes with a victory client so I can do things like get table rows give me rose from bakery table this is track energy that's all right okay thank you so I have a big very client I want to get rose from the curry table this is the native client so I'm doing this this is a really fat table this table has maybe couple hundred columns that contains all the metadata about our Spotify catalog this is actually real live production data let's you know let's see all right I get it iterator of my data so door next this is the untyped JSON you get from Vickery which is kind of annoying to work with obviously you're referred to columns is a using stream and you get object back obviously and also I can type so you can know it should be master matter later all right now you get object then you cut it to something that you think it is table I've done this like a million times but whatever now you get a nested record then I do that again track see it can be very annoying imagine running a pipeline and two hours later MPE somewhere deep inside your code so let's try it again by using a different approach I have this snippet copied so this is just an annotation the acree type say give me a tight from a query this is my query selecting all these columns the system already knows what columns I'm requesting from the sister from Vickery so it can do a dry run and figure out the result schema yeah I have a class but I don't have to write any code for it cuz why should I it generates classes and now have this track and AD cloth it's just too pretty string print it out you have you know fields with typed option for novel fields and let me get the cradle get typed rows back track entity I want to get my data in the form of track entity I don't need to re-enter the query again because it's already embedded in the type this data set is about 400 gig I think contains 70 million rows that's the number of tracks inside Spotify and they should take a couple of seconds down 10 seconds alright so I get a new air tracer next track anything it's a case card so are you attribute okay that's option but since I know it's sound and just kill it and temple yeah so that's how it works can easily pull data from bakery live and let's run a job again so I started this this repo without any parameter so by default it will run everything locally so let's run a word count text file so I already have this in my history so basically what I'm doing is a read this text file from the Google Cloud Storage IMAP split you know to lowercase come by the value and materialized which will give me a future at this point that the pipeline is not evaluated yet when I close my pipeline context submits the job to the Google Cloud actually this is running locally so it's not a mini and they just post data live from from the cloud and does the computation locally on my on my laptop so don't do it with largely that's a you'll blow up the JVM heap all right this usually takes like a second or two at work blame the network here but get my result back so remember I have a future and this is already completed so I can do a future wait for result value which is the value inside my future that abstraction its iterator you know take ten of them to this that's my work on what if I realize this is too big I have to run it in a cloud so I can do something like this say reconfig my operations with this Google cloud project some staging locations on bucket in a cloud and run with a real dataflow pipeline runner instead of the local runner I do that and give me a new context so if you've done spark each spark contacts or reports basically connected to a single cluster and you can do it live because then you have to tear down or expand or shrink the cluster in this case I can do that so I'm saying instead of doing that same code locally I'm just gonna copy paste same piece of code again and now instead of running locally I will submit a job to the cloud it actually builds a job locally and then submit a jar to the cloud and gives me a link our tracking URL so this week and I can see it so I start running we can take a look again it should take maybe two or three minutes let's go back to slides thank you so who's he using it it's Wi-Fi there are actually a lot of teams using it already most of them are new to Scala or she'll and actually a lot of them are just business analysts or data scientist who doesn't code as much but or you used to Python and odd but it's fairly easy to pick up we also try to adopt this open source by default model know that we're on a cloud so everything's open and github we have our own select channel internally for answering questions but we have open data and mailing lists so if you find some issue or have any discussion questions feel free to follow them on github or join the discussion and a lot of the features are actually driven by both internal and external users so they contributed a lot of the patches and feature enhancements - like the following components release rate that is one of the latest features already released it's a personal personalized new release recommendation like you know every Friday it gives you a playlist of all the new latest albums or track that you might like this runs fifty and one standard one Workers which is one call and four gigs of memory fairly small and consumes like 130 get data and then ingest them into two big table clusters we have one in Europe and way in the US there's a lot of the table mutations 110 million like one for each user the entire code is 129 so code yeah that's the entire job thing insights that's the the content creator facing side of the business so like if your label manager if your artist you want to see who's listening to my tracks from where we do a lot of slicing and dicing over many dimensions like track artists over different play contacts like from a playlist radio or recommendation geographical demographical dimensions and over daily monthly week this process like terabytes of job on a daily basis it used to be a hundred fifty Java MapReduce jobs now it's less than ten jobs because you can just do so much more in a single scholar pipeline and yeah master melody that's the the de metadata I was using in my in my demo that actually uses although scaling so the job might use from two to thirty five workers at any time based on throughput it joins 26 datasets from constant providers sequel Dom 6ml dumps and a lot of different sources 670 million records 120 gigs of data the code is 200 lines of code yeah and it was plotted from Java so the original code was 600 no more this is what a pipeline looks like in that web UI H this is massive but it's actually very easy to to write in Scala so this is like the number of inputs and the joints and group eyes and slice-and-dice happening inside a pipeline another little utility that we built to make people's life easier it's called big TV you know oh no what if does get into text files or any kind of files but we want to do data sets so this is what I call pairwise field level statistical div so you take two data sets given a give it a key function now you can have pairs of records from left hand side and right hand side the records can be ever be a curry probe of whatever and then it drills down into the fields so it computes like a delta you could be numerical it's just the difference between the two values for string it could be area distance we can override it we use that as a lot like latent vectors from machine learning so in that case it can be a cosine similarity or distance and now that you have a field level Delta of each pair of records you can compute it on a global level the global statistics min max average and also standard deviation those kind of things and a lot of the time there are like non-deterministic fields in your data like timestamps or hash code or whatever you can just ignore them or specify fields you know and also sometimes people use lists of strings or list of items as a set in our case sometimes a list of the countries of track is available you can specify that field is unsorted so you're ignore ordering when comparing the DIF you get statistics obviously the number of Records or pairs are the same different missing left or right hand side we use that to validate pipeline migration imagine if you have a Python map streaming MapReduce streaming job you want a rewriting she'll and they want to validate that it works exactly the same way obviously people don't write test when they when they write jobs in Python and then we have to figure out you know it's doing something insane and just give the result of the two pipelines and we can use that for sanity check in machine models like you to know parameter how different is my outcome is a statistically statistically significant or not you have all these step steps that you can use to to compute it the code you take the left hand side right hand side here on a function to auto join now you have a key and into options for the left hand side and right hand side value and if both of them are present you do your Delta and you end up with the output tuple that's the key the primary key of your data set the data that they altered the list of filters and also that the other type the second step is we map their scene into a map where the key of the map is your field field name it's like track IDE or you know use a age or whatever and then the value is just a tuple of all these different things but notice that all these things are actually some herbal have semigroups define on them so if you the first line of this snippet is eligible and we use the RG bullet heavily so we map everything into a tuple of semigroup summable types and then the actual computation is the last line it's that just some when you do that if someone's a semigroup over this map type the map of string 2 to the tuple and then recursively summons semigroups for every individual columns individual fields in the tuple to here yes so that's how it works other uses we use it for a lot of different other things as well maybe testing monetization user understanding ranking all sorts of things yeah finally let's talk about Scala or how it works a few interesting lessons learnt serialization it's the hardest part in Big Data there are two types of sterilizations happening in the pipeline you have to serialize your data a lot of the commons scholar libraries scouting spark storm they also they all use cryo and chill under the hood so crowd basically does reflection a lot of fights called manipulation to figure out how to serialize your your records or objects and she always a Twitter library that increments a lot of the serializers for scholar types dataflow requires the explicit coda implementation which means you have to increment your own for every new data type that you define that's annoying and because it's in Java there they abuse I sort of abuse a lot of the Java type system to make that happen in Scala we have craft X so you can do runtime a reflection to figure out what's in my pipeline and what's the best coder for that so that's data serialization and the next problem is a lambda T realization which is way harder and lambdas only serializable with Java civilization not any of the libraries that's available out there so that's annoying and your your lambda my pulling a closure that has fields or members that's not see realizable we use the closure cleaner from shield to basically a runtime nulling fields out to make them or for stun serializable and of obviously you can use the other tricks like making things serializable until they work and marking fields it's transient lazy we all until they they work so those are the type of tricks we have to do to make sure the pipe lines as runnable in the cloud repo another pretty challenging topic in spark the repo actually transports your Lunda bytecode from the master node to all the workers and there's a little HTTP service on the workers to to download them data flow as i mentioned earlier it's a it's a static it's a huge remodel so you have to have a jar available while submitting a job and once you submit a jar to the cloud it's it's over like there's no communication between a local node and in a cloud so basically what we did was we had to wrap all have our own custom class loader and i loop and then in intercepts the compiled bytecode put them in the job jar and submit it and the time who know they used to manage the hadoop clusters qualify so his previous job was to restart a name now like every other day you know of the night but since now we don't have to operate our own own cluster he did all these cool stuff he's actually over there somewhere Rob yeah alright a map using IntelliJ another challenging problem so if you if you guys ever use scenes like shapeless or or simulacrum or some other scholar libraries that generate code compile-time some of them are not what most of them are not picked up by IntelliJ so you might see red underlying stuff IntelliJ that's actually compatible in SBT the the program is in college it doesn't kick up Marco generated code and liking are okay is the bakery case classes so obviously the solution is to write your own SVD plug-in that seems like a mature thing that every tech company is doing these days just write your own SVD plug-in so we did that and it picks up macro generated classes for IntelliJ so now you can do like tab completion of your data type in studying geology a shield for Apache Japanese Japanese the web UI for originally for spark they also runs on a bunch of other big data frameworks including BigTable so another nice thing about geo or dataflow with Japanese is I demoed before in a command line repo there's no cluster you need don't need to set up a cluster so in this case we can just distribute run about gr to the users they can run everything on a log laptop and it only talks to or communicate to the cloud when you perform some real computation like submitting a bakery or or JA so that's another potential use case for or this sapling interpreter for shield and a couple of other experimental features playing around with shapeless parents protobuf some of our users prefer protobuf over a bro and if you've compared the two there's no way to manipulate schemas in a generic way in protobuf so we kind of worked around it and now you can deal with this sort of like a pro so you have utilities that can inspect throat protobuf serialized fights and deserialize them into Jason and vice-versa so it's mostly for for productivity rather than in performance so yeah lassy what's next by default data flows across both streaming batch so she all supports both also by default the one constraint in data flow is that to update a pipeline running pipeline or the transformation needs to have deterministic naming which is not a case in Scala because we named them the functions and and the file names in your source code for for the bug bug fixing or troubleshooting so that's something we need to work on right now we have a working branch that runs Apache being zero to zero and then we're walking on the rope 3.0 the latest release obviously it's tightly coupled with dataflow right now but it should shouldn't be too hard to dig up all those components and make sure it runs down safe link or spark as well in the last thing once being a party being stabilisers it's our plan to donate it to Apache as part of the beam Scala DSL that's yeah that's it that's the end of my table so if you have any questions we can do it offline I think we're running out of time yeah [Applause]