Spark and Protocol Buffers - An Awesome...

scala.bythebay.io: Nadav Samet, Spark and Protocol Buffers - An Awesome Combination

Recording: scala.bythebay.io: Nadav Samet, Spark and Protocol Buffers - An Awesome Combination

good morning everyone my name is Nadav summit I made the CTO and one of the cofounders of the company called to Accord it's a silicon valley-based fin tech startup that using Scala we fall to Accord I spend many years at Google walking on a various projects like Gmail Google Apps and Google travel and like everybody else at Google I spent a lot of time writing and using protocol buffers so in today's talk what we're gonna do a we're gonna do a gentle introduction for a footballer buffers I'm gonna take you from a beginner to an expert in seven minutes and they it's really possible and I'm gonna show you how a little library called Scala PB makes it really easy to use particle buffers in Scala and why you should care and why you should use protocol buffers and at the end we're gonna have a live demo that's gonna show you a full app using protocol buffers all the way from a scholar jes web app to a play for a mock web server to a spark streaming aggregation server and this is gonna be a really amazing demo I've never done anything like this before it might fail miserably so we're going to find out together so it was a story that they may sound familiar to you and we have it we had it a to a code when you just start a a project a lot of business requirements keep coming in that keep changing your data model so for example in our case we had users and we had the counts and in the beginning every user had exactly one account but over time we discovered that some users need to share a single account and sometimes say there is one account that there's a there are users that need to have multiple accounts as every time that those changes came a we followed by by rolling out sequel migrations and as you all know rolling out sequel migrations to a live distributed system is a pain in the ass that's a technical term so a the idea was to move away from a form a storing columns in my sequel that represent individual values and use a key-value store to store serialized versions of a document stole all entities as values in a key value store em so let's zoom out a little bit and talk about the protocol buffers which is the approach that we're using to a to store a our entities so how many people know what protocol buffers are very good right almost everybody how many people use them in a Scala few right about half people in the room so cool I'm gonna give some introduction and you know that most of you already seen some a polar buffers at some level I think it's still gonna be interesting to look at some of the of this angle so let's start with just defining what protocol buffers are so protocol buffers is a mechanism that was developed at Google to represent a structured data messages like case classes and it comes to the specification also says a dictate how those messages are going to be serialized as a as bytes and now they're going to be parsed from a byte stream back to to the to the messages themselves and it also ships with a compiler that gives you a generates code for you to generate serializers and deserialize errs for you in Java C++ and Python but unfortunately not in Scala so I had to write Scala PB so a 1-1 cool thing before Before we jump into what protocol buffers look like you can already see is that the beauty here is that the language is that the the polar buffer language is agnostic of your platform so you can serialize message in Python and read it in Scala or the other way around between basically any pair of supported languages so here is the what the protocol buffer schema looks like we're using in this talk syntax a equals port of three with it's the latest pacification of the protocol buffer language and this is the an example of a message called person it has a first name which is a string a last name which is a string a name an age which is a number and every field as a uniquely unique identifying number with it so once you have this schema you can use it the Potosi compiler which is a built by Google and that generates the serializers and DC realizes for you in in various languages and if you're using Scala P being in Scala you can basically ignore all of that and just use SBT compiler to generate the code for you and compiled in a single state single step and the that's how it's kind of a integrated in your development flow if you go back and change the protocol before you just do SVT compile as normal so you have this protocol buffer and a scar PB would generate for your case class for it so you basically get the same thing just in Scala and you deal with it like any normal case class you just say Anish initialize it and they put some values for each field and then you can call to byte array and that would give you a stream of bytes this is the basically the binary representation of this message which is very compact only something like 15 bytes for this one and you can also do the opposite you can start from a series of bytes from a stream of bytes you can get it from the network or from disk or from your database and call person dot parse from which is a method that is automatically generated for you and it will give you back the original fields that you got in the original case class and another cool feature of protocol buffers that make it that that is that makes it really good is that you can create complex messages by composing simple ones so for example here we have an address message which has a street and city fields and a bank account message which has a routing an account number and then we can make a person message that there's an address field the others that we just defined above and a list of bank accounts and then you can just if you just told you this person in your database or use it in your data processing pipeline then you don't need to deal with those many too many fields that you'd normally have like you know the relation table that you'd normally see in my sequel and this is the generated code for that so basically what you'd expect you get the case class for the address the case class for the bank account and a case class for the person and the and the person has the Eddas field which is an optional address and the a list of accounts which defaults to the empty list now cool thing to know especially if you already familiar with protocol buffers is kind of like dive deep into the wire format which really helps you understand why why certain things are they possible with protocol buffers so if you have this a person message and we initialize it with those specific values Joan's meat and age 34 and we serialize it what you actually get on the wire is something that looks like this basically we have the field number followed by a value field number followed by a value and another field number followed by a value a so if you think about it you basically see how we be if you compare it to something like JSON where we will under wire you send a lot of punctuation and the field name itself like you know with the in quotes and all that you know we only send a binary representation of the field number and beyond the saving the cool thing about it is that you can rename your fields over time so let's say you discover that you are the typo or you want to change the terminology that you're using in your schema you can just change it deploy that server and it can still exchange messages with the old servers that still have the type or the older name and the system can keep talking with each other because the wild format the binary representation is not impacted by by the rename a another cool feature that is very useful if you're doing rich data and a lot of business logic with the protocol buffers is they using enums instead of strings so for example here we have a bank account protocol buffer and we want to have some indication of which account type it is a checking account or savings account so we define an enum and then we can refer to that enemy in the bank account message and if that leads to a very dramatic Scala code so when that get generated we get a sill trait for the account account type which is basically the base class for a for the account it has two predicates is checking any savings which are boolean and you get two case object for each possible value of that innum and a the cool thing about it that it leads to a very nice usage when you actually write your business logic and say you have a big bank account you can do B dot a account type is checking and do some logic if it's a checking account as to something else and moreover you can use a pattern matching and they run a case for every writer case for every possible value and a the cool thing about that is that beyond that it's very dramatic Scala and it's so much better than comparing with that you know how to coded strings for every possible account type if you use something like JSON is that the compiler will actually save you so if you ever somebody ever adds a new account type and you recompile the compiler will tell you give you a warning for every every pattern match that's not exhaustive that you haven't specify what to do so the compiler does really help you to to write code that is they're gonna and all the cases and it's really in our experience it's been great to have a system that evolves and gets more more complicated over time by using by you creating a lot of structured data like this so now that we have protocol buffer experts let's say compare that to just summarize what we just said and they compared that to a simpler more common JSON approach so a we have an efficient binary format every every key every value is stored in its tordesillas in a compact binary representation it's easier to programmatically access the data so you're basically walking with case classes and the idiomatic scala rather than you know in the JSON case you're not if you have read to walk with JSON it all the time feels that you're navigating a tree of strings going down to get the value you want and you actually know what to expect you know what to get in every in a in every field you know which type a you're gonna find so a you get the type safe in a process communication so if you have two servers that are talking protocol buffers with each other and the two servers are are and know what type should be in every field they know the format of the message there are no problems at runtime that you like you'd get with the modem of the dynamic formats what I found really really beneficial is when your project goes and grows and you have a front-end team and a back-end team and a data science team and they all have they can all reference the same protocol buffer schema and they develop against that so there's a single place that all the teams can look at and there's a documentation what fields what fields mean what is the semantic meaning of every field or what like the special values for that field what does it mean if that number is -1 what it means that if this message is absent and so on so creating this single place where the entire organization and tile project even if it's divided to multiple teams it's been proven to be very helpful for us in a scaling our business logic and the system grow more and more complicated and the last benefit for protocol buffers it's great for evolving schemas it's so easy to add more fields or remove fields and make changes without without breaking your running system so let's take a look at the at one example so let's say you have this a personal message that we keep referring to and then you stole it you notice that you stole the Aged and then you realize that it's not a very good idea to store the edge of the person in a on your database because the edge keeps the person's keep going up and they what you really want to store is the date of birth or the year they were born so no problem you basically just create a new protocol buffer if you basically add a year of birth field to a protocol buffer and in your new in the server to just deploy they can still read the old field but they have access to the new one which it would be 0 by default and then you can basically provide a function that would a that would try to get the age from the year of birth but if it's unset if it's 0 it would fall back to the age and over time what you can do is to write a data migration basically a simple script that would populate the new field and eventually drop the old one and it's because it's so easy to add on a move with those operations can just be hauled to a live distributed system without any downtime this is just an example of like a toy a toy a schema evolution but you can do a lot of stuff you can add fields and remove fields like we just mentioned everything is optional so the parcel of a if you remove a field and you try to parse something and you get the old version of a message at the head that original field your past won't crash it will just drop it and you can rename fields like we explained before and you can convert between comparable types you can convert between in 32 to in 64 you can convert it back if you don't mind losing a few digit you can convert between strings to byte arrays for byte arrays back to string if there you tf8 there are few rules but it's you can do a lot of stuff and you can also convert between a single field to a repeated field and back I want to talk a little bit about the motivation for writing Scala PB so I protocol buffers ship with the with Java support and because they're in java and they don't have case classes they use the builders and this is a little bit painful to do especially when you have nested structures so basically you get a person builder and you set a few fields and then you get need to get a nested builder to set the nested fields and then you call builder dot build and this is get it gets very cumbersome when you have a few few layers of depth over there you need to store them the builders in temporary variables and so it really nice another thing is that they started to build their API before Java got the optional field before Java 8 so you they provide this as as address or as whatever methods to let you know if a field is present in a message and and this leads to a code that you basically all the time try to lift those fields to options so this is what made me a start Scala PB and the goal was to make it easy and idiomatic to use protocol buffers in Scala I want to use case classes I want a development flow to be integrated with SBT so I can easily change my protocol buffers and not worry about distributing proto sit or all the developers and I want to make it easy to make updates to nested fields and when by the time we created Scala PB we already had tons of code with the java protocol buffers and it was really important to be able to you know nobody's going to move from scala to java from the java protocol buffers to scala overnight you want to provide a way to make it gradual so a scallop it we can generate conversions between the Java protocol buffers and the scallop protocol buffers and back so you can basically convert part of your system to the scallop protocol buffers and basically have those converters as a barrier between the old code and the new code and eventually you move more and more code to the other side to the scholar side you can actually use you can actually to do the other the opposite direction but they don't recommend that so a but and then as the project got more and more successful and we got users asking told us they're using Scala protocol uses Scala PB in name in all sorts of contexts that we are we weren't using at the time like in SPARC and in for akka and G OPC and it was clear that if we want to make this project successful we've gotta help users a wall proactively in those environments so that made that we invested a lot of time in creating adapters creating documentation for those environments and the some users started to contribute so some user just one day sent me a a pull request for a gr PC support for protocol buffers so thanks Kenji if you're watching this and beyond that what is a what's a we also learned is that Scala users really like to have a lot of control over the generated code so we provide a lot of tweaks a way to tweak and lot of knobs to control the types and certain aspect of generated code for example a we allow you to specify custom types so if you have this a let's say that you have this click event that you're using to track clicks on your onion on your web app and it's very common to you to store the time as a in 64 just some representation of milliseconds since epoch and a however when you're doing this you know some higher-level codes and data analysis you usually want to work with the higher-level type like some more something like daytime so you can spell tells copy be to use a shorter time daytime a when you walk with it on on on the Scala code so Scala PB will just take that click event and if the caseless would have a date a the daytime field for the timestamp and the cool thing about it is that this is completely transparent to you as a user when you have this case class and you serialize it it would be converted today to a 20 in 64 and sent over the wire when you get a message no matter what we encoded it by in Python or Java it would take the number and convert it back to a to a day time and all that magic happens say all you need to do to make a make that magic happen is to provide an implicit a value called type map where that tells Scala PB how to convert between your type and the and in the original type the last thing I want to talk about on that scallop PB provides is a is nested updates so this is a classic problems with classic problem with the case classes in immutability so let's say you have an a that has a member B you have a class b that has a member c of type c c as a member D of type D and D finally as AE of which is an integer and you are giving this a from the outside and all you want to do is to change the e that's kind of nested deep inside it you want to set it to 17 so it is the code that you normally need to write you need to take a and make a copy of it and that you need to have a copy of a dot B which means to have a copy of a dot B dot C which in to have a copy of a dot B dot C dot d with a sign 217 obviously you don't you never want to write something like this this is not fun right and this is especially infuriating if you compare it to what the guys are doing 30 program in can do right so with Scala PB you get e you get it set you can use this technique called Lenz's and that lets you do those mutations in a very elegant syntax and scalloping bishops with the small lenses library that lets you do a you know nested assignments but also a transformations like calling to uppercase on on a nested field or concatenating to a list and by in this way you don't have to you know to make it those cumbersome copies and also even in this example were scrubbing sensitive data who is coming account numbers by iterating over a list and signing to each something to each member of a field so this is a pretty cool way and if you try to add it with copies it's it's really you know it's not fun so a now demo time so the demo is that I'm gonna do I thought you know it would be nice to do a survey or which programming languages or favorite language but because this is a Scala scale by the way I kind of can get guess what the answer is gonna be so I thought it would be interesting to actually you know that we vote on our least favorite language I think that would be way more fun and to make it even more interesting a the stuff is also going to include a your age so we can see what is the least favorite language but for each age group and I think it's gonna be really interesting to do that so what we're going to build is something like this we're going to define a protocol buffer for that gonna represent the vote and we're gonna have a Scala GS client that's gonna render a UI for you that you can vote from you can you'll be able to vote from your mobile phone or your laptop and I'll give you the URL in a bit and this would go to a place where mock web server and which gonna deposit the votes faithfully in Kafka which is going to be our message bus we're gonna have a spark streaming server that's gonna a subscribe to the to the topic in Kafka and gonna aggregate the vote put them in Redis so play can again show them to us in in real time on a dashboard with in Scala yes so can we do it in kind of 10 minutes that's right so all right so yeah I'll leave the help here with the mica so I'm gonna switch I'm gonna switch to IntelliJ okay so you know I'm heading I'm editing the the protocol buffer that we're going to have so a let's write a message called vote and we're going to define an enum for the language and let's say have some value for unknown which you know deserved for a default value that we have to provide and I want to hear from the crowd which language you want to see here what assembler simply write a what else Erlang what else Lisp or Lisp what else - okay one more let's have one more Pascal Pascal has Co okay let's say that's gonna be loaded okay so a a now I just defined the type now I need to have a field that's actually going to have this type and we said that we're going to also take your age so here we go and now I I'm going to go to my shell and I'm going to compile a despotic or buffer good morning SBT I'm going to also start the server in the meantime let's see if we already have the protocol buffer generated for us yes okay so we have a final case class vote and it has a language field and let's click on this we see the same trait with all the languages that we just chosen as predicates here and basically we see here the case objects that were generated for that so pretty cool and what I also prepared in advance is the ecology sui I'm not going I'm not I would not dare do ecology as in kind of like in in front of a live audience so so it's kind of risky right so a so basically what we have we're using the awesome a scale attacks library to generate the UI at HTML we have form and then we basically iterate over the pot all the possible language values a filtering out the unknown language and they were ilding a div that has a radio button for every language and when the form gets submitted we are basically serializing it to a byte array and they're making an XML HTTP request to the backend so let's try it out okay so I'm gonna give my answer not really but let's try a so not implemented and the reason is that basically this answer goes to the two-day back-end which doesn't have an implementation for this a for this action so let's look at the hood controller and it is the submitter method and here we tell play to basically take the body and implement it as a protocol buffer so we can have a V vote equals request dot body sorry and we can do V dot language match and have a IntelliJ Elpis you know complete the code for all the languages so it's really nice you know that there come the ID can help us because it's all very structured but let's do this if the language is I if the language is unknown we're going to give a bad request pick a language else if you are if the age is invalid so age or you're too old let's say a invalid age and then else we are going to say we're going to say thank you a let's try that so here's my conflict one minute for processing a I'm going to give a language but not an age invalid age no a language big language but if I pick both thank you so the last thing I'm going to do is I'm going to tell it to if the vote is valid I'm going to send it to Kafka and basically this is the vote is the topic name and I'm sterilizing it to byte array and let's say deploy it to our to my server and yeah looks like it's starting and in the meantime I'm going to show you how the votes are going to be aggregated so we have a spark streaming a web application application here that subscribe to Kafka and gets all the votes as bytes the next thing it's doing it's a parsing those votes from bytes back to case classes back to protocol buffers then it's mapping every vote to a key value pair where the key is a tuple it's the language and the age group how do I get the age group I divide by 10 and multiply by 10 I basically rounding it down to the closest multiple of 10 and so basically all the votes now I hid our kid by the language and the age group and I'm using a function called update state by key so I'm basically on a count how many vote where I have for each key to a language and an age group so the fact this the applet said by key gets a parameter which is a function which is a a new sequence of vote that has the same key so they are basically need to be counted together and the old count which called current start which maybe it's an option of an integer so the first thing we do is get all l0 of it so that's the default value of 0 and we add votes that size this is the list that I'm getting of the vote that have the same key and finally once I have this aggregation I write a little bit of code to stow it in Redis that that's all there is to it so well a looks like we have a like I need to do this again it's a complex demo I won't you so and in the meantime a it would be great if you could take your phones out and they go to a demo dot the summit calm and I can show I going to give you a QR code it's going to be D here in a second here we go so you can scan the QR code if you want to get URL it's also written below the votes are coming in and in the meantime I can also take questions yes okay thanks for the question so the question is how do we deal with the required fields in four to three when basically they're out of this pack is produce still gonna stay around and I guess that's kind of the gist of it so yeah so potahto as far as I understand is gonna stay around there's tons of protocol buffers out there and tons of users that are still using portal 2 and there was no really a very good reason to break them and so all the time when you know when we use port - there was this notion of a don't use required fields they're evil they basically make it very hard to do those migrations because you can't remove them from your passes so this is basically something that you know the push was always to get your a your validation code to be outside the protocol buffers library and they have your application validate if the if a field that you really want is in in the message and not rely on the on the parsers and serializers to to tell you that your message is valid or not am in Porto 3 basically you come from you lose this option and and it's a little bit worse for some users because all the primitive values are default to zeros on empty strings so they lose this even distinction between none and some zero right so there is a solution for that at that Google is shifting called well-known types which is basically wrapping all the prima tubes inside messages and the messages can be in a they are really state still optional and then again you have to use your you know application logic to to enforce of they're there yeah more questions yeah so if Paulo buffers you basically if you're using Scala GS for exact uestion it was how do you a I'll do you what happens to debugging your your your server if your front-end is using protocol box if your wire format is binary basically because in Jason you can use the developer tools and click around and see so yeah this is true this is basically where there is a missing tool I'd like if somebody would just write a Chrome extension that can pass particle buffers over the network I don't think it's hard to write it and I however the reason something like that and we resort to printer Lance when when we need to debug things I couldn't hear you so for any support we do have the yes so basically any is a tie so the question is what we do with any support the the any is a type that basically it's kind of a dynamic variable that you basically get a it's a message with two fields the type that it's supposed to be and think the name a byte array and the byte array is interpreted in different ways based on the on on some URL that you giving it represent a message so we have some support in Scala PB for that if the message is known at runtime you can basically get you can there is an implicit a conversion available on any that can give you the the actual type but you need to know what to expect I think there's it involves using as instance off and it's not great but that's what we can do right at the one off you mean yes yes there is something called one off which is a it's a nice a it's a nice feature of worker buffers how many minutes do we have five cool so I can show I have a backup slide for this people are interested yeah so here for example it's an it's not a way to do it the protocol buffs give you a very powerful a way to structure your data so for example you have a payment event and you want to store we want you want to say how the payment has been weighed so you might have a message a credit card and we have the bank account that we define in an earlier slide and then you can ever one-off payment type which is like a union it's either this or that but not both on but it could also mean none of them so you have a payment type with a bank account or a credit card and the scalloping would also give you a silt rate and two case classes that extend each of these options which is it's really cool to write logic like that but this is a way to basically have the parcel give you a little bit more you know validation before the messages ended to your business logic yeah the question is what what problems we had walking with protocol buffers and I can't think of any actually it's a great question so I I you know the one thing I'd say that you know when you write something like this it makes it really hard for somebody like let's say you have a back-end and front-end speaking skull a protocol buffers and you just wanna send like if it was Jason maybe you just push like Andals Cole Cole oh he didn't feel that nobody knows what it does except two guys that are implementing it and you can't just push random stuff with your poker buff it's basically rigid and it's locked to that schema so it could be good or bad thing but basically it doesn't allow you to kind of like throw temporary variables inside it for some head doc use cases like you know the color of the UI or whatever so sometimes it kind of like stops you but maybe in the long run it's a good thing yeah yes yes this is basically a you know here we use this colored what yeah the question we is whether to is possible to use a protocol buffer with angularjs and other font and technologies so I don't have a lot of experience of Mitch mixing javascript with with Scala GS however there are there is a protocol buffer code generator for JavaScript that Google provides that can basically if you already have your fountain in JavaScript you can use that yeah yeah so the it so the question is how protocol buffers compared to Avro so it you know it's very similar the you know the small common thing that are different about tends to be more dynamic so the schema is a also I believe included with with your documents also and and it's more symbolic so you basically don't have this say when a message is represented you basically I think it leads to I don't with a lot of experience with Avro but I believe that things like rename are resolved symbolically at one time rather than at the wire format yes yeah yeah cool a yeah I'm saying that bash is getting a lot of eight from users to any 240 but I feel see like a lot of old people in the room a like under than ten people that don't like assembly and I can understand that so way cool more questions yes okay so the question is about Jason conversions for protocol buffers which is something we use we still use a lot a basically you can convert a protocol buffer to a JSON and back and you can then then this gives you the the option to write your front end with the normal JavaScript technologies without using Scala GS it works pretty well like we use it all the time yeah a you can do it also with the there is a scalloped eb jason 4's a library that we wrote that lets you do it for any any version yeah I mean come again yeah so basically you you have a method set write them to an output stream that's kind of the easiest way to get them to anything yeah yeah yeah as long as you have sometimes can speak the outputs the abstraction which is probably everything then yeah in the back you can use case classes directly it's Park inspectors nothing that you need to do you might want to enable a cryo a civilization that's more efficient and there's actually I'm gonna post a Center on how to do that so there's a way to do it so that's all the time we have I'm gonna be hanging around here if anybody wants to ask me any question offline thank you [Applause]