BDSBTB 2015: Dan Osipov, Scala at Shazam

BDSBTB 2015: Dan Osipov, Scala at Shazam

Recording: BDSBTB 2015: Dan Osipov, Scala at Shazam

so for those of you who did not raise their hands Shazam is a really cool app you should get it it's on that play store on the app store it's real awesome on windows as well so when I meet people date and I tell them yeah i work at Shazam they typically ask me like oh like how does it work like a press a button and it tells me what song i'm listening to like cut how does it do that and yeah that's a big secret that's one of our recognition clusters yeah but I'm actually here to talk about how we work with data and to give you like a high level understanding of our infrastructure this is kind of like a partial view of what our data architecture looks like essentially we start with events generated by the mobile app we store them in s3 we do everything in AWS and then we use scalding to process those events and store them in dynamo red shift back to s3 send them to like the rest api is whatever so scalding has kind of become our central to our data processing and that's the what this talk is about so what is scalding scalding is a scholar DSL on top of cascading cascading in turn as a java library that compiles to MapReduce jobs that one on Hadoop Martin mentioned the features of Scala that make it ideal for dsl this morning and that's that's essentially that applies equally well to spark as well as the scalding scouting has been was developed originally at Twitter it was open source there and they still contribute fairly actively to it the jobs that are written in scalding ultimately compiled down to MapReduce latest version of cascading works with other execution engines specifically like with Taz and there are rumors of potentially using spark or flink as a back-end execution engine the key abstractions to understanding scalding are the sources pipes and sings data flows from the sources through pipes and gets dumped into things the way you interact with these abstractions is through two api's one is called the fields API then you will know what this tick line and tick word are call it out symbols yeah there are actually a core feature of scholar but they're not actively used actually scalding is one of the few places where I've seen them used the fields API is deprecated you should not be using it I'll instead be focusing on the typed API this is a skull conference I don't have two cell types to you this is just way better and like I said the other ones not it still works but it's not actively developed and although kind of like development resources are going into the typed API it can use tuples it can also use case classes there are macros to get it to read into case classes that are really cool when they're kind of advanced feature of scalding the pipes have a bunch of operations not going to go through them but essentially they just look like scholar collections you have you no filter you have flat map you have map all the stuff you expect on a scale collection and even more there are a number of joint implementations where they come in handy is if you know a relative size of one of your pipes versus another you can kind of hint to scalding and it will run the joint in an optimal way right so for example like if you join with a tiny pipe that tiny pipe is going to be replicated to all your map or nodes and that's going to be much more efficient testing is something that often gets ignored in data processing it's actually really important and it's it's a first class citizen and scalding so if you have a bad job running regularly like you want to know that it can handle whatever input you want to throw it right and where this really comes in and becomes handy is when you have to add features down the line like half a year after you originally written it and you don't remember what it is right testing gives you a simple input sample output and just verifies that everything runs correctly the sculling tests they run our standard scholar test tools and just specify the arguments then for any sources you specify where to get the data and you do all the verification in your sink the one challenge is in a scalding job all your kind of setup goes into the job constructor that means it gets really hard to unit test you can use a reflection to kind of get around it or what I frequently do is in my job I just mix in a trait and then I can instantiate that trade and run my unit tests on the trade just separately without scalding so so this is a big data conference so I'm contractually obligated to share your word count example otherwise Alexi will take away my Big Data lessons I'm going to go through a more involved example and I'll call that like a word histogram essentially what we want to do is see what percentage of the word of all the words in a document are represented by specific word and I'm going to run this on song lyrics so to start your job class is going to extend from the scalding job the arguments is what gets provided on a command line I use them for paths essentially where I want to get the data and this is the first pipe these are kind of like the guest upwards you know like words that are frequently occurring and that we don't care to count so I load the Manas text it's just one word per line create a type two pipe from them and then group them by identity I do the group because I want to join this pipe in later so that becomes important later now we load the actual document the song lyrics so we take the input tokenize it so now at this point we have a type of individual words group them by identity and then do the left join so the way to think about the left join is if you think about a left join in a kind of like a relational table you have all your values from the left table and then would whichever values from the right table match on some key you get those columns there but if they don't match you just get Knowles acceptance collar we heat nulls right so we use option so when i do the flat map on the values what i get is a to bowl of the word from the left table and then an option of the word from the right table in this case if the option if the option is some that means i have a stop word in my right table so I just want to ignore it and remove it from further processing if I get a nun that means this is not a frequently appearing bored so I just wrap it in an option and attached to one for counting and because i'm doing the flat map this option essentially goes away all the nuns goes away so at this point i am left with it's still a grouped pipe but it contains the two poles of the word and the count and i don't care about the group key so I just kept the values all right now to get the total number of words without the ignored words just take that count and sum them up and I get the total pipe it's still a pipe it's not an actual value because what we're doing here is we're kind of setting up computation then I take the all the words group them by the word again take their size so count them up convert them to pipe then get the percentage that the word represents of the total by using this map with values so the total pipe is a value pipe essentially it's a pipe that contains a single value single number so this is actually joined but it's a really efficient joint because that value can just go to all the mapper notes and then at this point we're done but I only care about the top 10 most frequently occurring words so I just order them take the 10 and write them to the output I ran this on some top 5 Led Zeppelin songs and top 5 Justin Bieber songs and here's some results before go on any questions is anyone like thoroughly confused make sense cool this is going to be short presentation you guys alright so back in the mapreduce days on the way that Google described MapReduce it's very kind of a classic you know scatter gather algorithm right but the way it's always implemented is there's this intermediate stage essentially if you have a mapper that has to rely on the reducer to do the counting that means the mapper has to send it to pull of oh I saw the word and once oh I saw a word and once I saw a word and once multiple times and that gets really bandwidth heavy on the network so Hadoop has introduced the notion of a combiner which essentially runs a mini reducer in the mapper note takes those tuples of and some of those some open says to the reducer hey I saw and you know 55 million times and that's an optimization to skating and scalding they did kind of similar it's a kind of like a map site aggregation not necessarily combiner but it works in the same way and it makes sense for counts right but are there any other structures that this works for and it turns out yes as long as you have emerged operations you can represent your data structure in a way that scalding will be able to do this optimization on the website so the merge operation is present in these things called semigroups if you have a zero you get a mono it and a bunch of them are available in the library called algebra and I listed some of the kind of like the not frequent occurring once here you get them for essentially all the primitive types for things like sets maps list a butt out of these like hyper log log turns out to be really effective for doing distinct counts right it's a approximate algorithm but if you have to do a unique count it's very effective and nothing is like when you start working with scalding like you have no idea that there's decent 10 it's on to the hood and you have no idea what those are but then one day you have to look at the some method and it takes like an implicit mano it and you just look like woody here or there its realization is another kind of like a cool optimization on the latest scalding release what this does is it it can speed up your scalding job by around forty percent so when a MapReduce job needs to shuffle data the records they get collected at the mapper and uncivilized center the reducer we do sir deserializer some swords them sterilizes them in writing to disk so if it's able to do this sort on just the incoming bites without doing that dear serialization serialization you save a lot of CPU and that's what this feature does essentially you have to define this macro pull it in and yeah it does it for you as long as you're used like primitive types or case classes it will generate all the code that's needed under the hood so what is it like to write a scalding job the way I do it as I typically write a test first in looking like a standard TDD fashion I set up my sources my sinks those are typically defined by the Job Description can help by the problem and then let the types guy do essentially like I use IntelliJ and I get code completion so when I type a dot on that pipe gives me all the methods I can use and I look for the one that I want to get out of that pipe it's a very effective way to work so this is a really simple recommendations algorithm it's not going to win email inbox but it does demonstrate a couple of interesting scalding features we we start with like a input of a user an item and a rating so user has rated some item gave it a like of 1 over 5 we create a matrix out of them so this uses the scalding matrix API if you have a type in this case its input tuple is a it has a type of for the row for the column and for the value as long as you have those three you can create a matrix out of it do the normalization get the edges between items and then use the dim sum cosine similarity which is like an approximate Cosenza Marty and there is a blog post from Twitter that describes this in more detail it's also one of those optimization things when you have like millions of items to compare it cuts down the CPU use and runs your job much faster and yeah finally like unfolded remove self-similarity and just write it out so the basic recommendation algorithm in like four steps and so sometimes use the types to advantage can anyone tell me what's in this pipe two strings but do you know what they are I don't know so what if I tell you this it's it's the same code essentially but this type Elias just use it for documentation because now I can read this I I know exactly what's in this pipe and I don't have to remember always the account ID first there's a song ID first this is much more clear you can also use a debug operation on a pipe so when you have a sequence of these operations they get and along insert the debug in the middle it will print out the contents of the pipe at that point so after its run the mapper but before it ran the flat map if it's a lot of output you can limit it by providing this kind of like a configuration object told where the output goes tell it how often to print the two balls if it needs to print the column names looks like that sometimes you actually want to see what the end dag will be created by cascading each of these is like a MapReduce stage and give you can generate this file from by using the scalding tool and passing it the tool graph parameter normally you would run your job through the scalding tool but if this parameter is present present it will not run the job it will do like a dry run compile it and write out this file which you can feed to graph ways and just look at it etsy has also released the gvm profiler for Hadoop and specifically for scalding it attaches a little agent to each one of your mappers and reducers and generates this flame graph as long in addition to bunch of other metrics singing old familiar with flame graphs okay so I better explain it so what you have here is a naxx access you have time and on the y-axis you have the stack depth and you can trade this but all of these are like class paths and the method names so it's each horizontal bar is whatever method is executing at this time so you can imagine like the lowest level is your main java method that gets executed and then as it calls in deeper and deeper into the stack you go deep and then these methods actually don't execute for too long they just execute and terminate and where you have kind of long bars that means there is a lot of time spent in there so this one is like decompressing the input stream so presumably it's reading the data from desk and doing the decompression and then the rest is like your computation and it that JVM profiler it can write the all the data to generate this flame graph right to like graphite or influence DB it does require a little bit of setup so concurrent which is a company behind cascading they release a product called driven and it generates this kind of a map of your test computations this deck and also monitors you're running job and prints out some counters and lets you see where potential bottlenecks in your computation could be so you've heard about spark enough of this conference so you might be asking like how is it different from scalding like Glenn would you use one versus another so I gave you like word count examples but there are API is essentially the same even for non-trivial jobs you can translate from one to another I typically find that scalding jobs tend to be more stable especially given with like sterilization and memory handling spark obviously has like a male lab and graphics and streaming and sparse eagle and bunch of other libraries so your choice is going to be driven by whatever problem you're solving a spark gets talked a lot about for its performance and it's definitely faster scalding on test is supposedly just as fast and deploying scalding today is pretty much much easier than deploying spark especially if you have a existing Hadoop cluster but that's changing really really fast yeah on this point there is a great post from at sea by I believed Emma Kinley that's a titled choose boring technology where he argues that us developers we get attracted to shiny things and we jump on the latest greatest technology but for some problems it actually makes sense to use something that's maybe more boring more proven and just to get the job done with that which is a misfiring this is interesting let me now talk to me and any questions so the question is about format we use for storing what we ingest is Jason because that's easiest for apps to send and then from there we transcode it into a bunch of formats and then feed it through our pipelines so we use a bunch of stuff within our pipelines if we read in Jason we quickly transform it into case classes and tuples if we read an avro then kind of translates into a case class automatically sing with parquet oh so the in installing their macros to generate case classes from schemas for parquet specifically and I think there's there might be one forever as well not sure yep we've tested that out we use it for some things like I said you know when faced with a problem I if it doesn't require like having machine learning I just go with scalding it's proven it works there's a lot of infrastructure and also with deployment like we are on AWS and spark wasn't supported by AWS until recently it was possible to install it using like bootstrap scripts it was just a pain no nose cool um I'm around to rest of the conference so if you guys have any other questions find me Thanks