SF Scala, Rolando Manrique, fs2-blobstore Store implementations and use cases

SF Scala, Rolando Manrique, fs2-blobstore Store implementations and use cases

Recording: SF Scala, Rolando Manrique, fs2-blobstore Store implementations and use cases

[Music] okay hi everybody my name is Rolando Manrique I'm a software engineer here at land up I joined recently just like January this year and yeah let's talk a little ops tour this is a library that we open-source a couple of months ago so it was a great learning experience and I think it's interesting to look at the the implementations the storm recommendations that we ended up doing and with fs2 and and some of the use cases where we're using this first of all let's get the disclaimer out of the way so lawyers are happy opinions are my own not land of official position but yeah having said that let's definitely talk a little bit about land up as you can see here and the words behind the screen also say on the wall Landis wishes to provide anyone with a path to better financial health and one of the things that are really that I really like in this short amount of time in lender is that this doesn't come this really come up all the time in our meetings it's like it's not just words in a wall it's actually we ask ourselves in meetings is this really the best thing we can do for our users and this is this how we can help them get to a better financial position so yeah our products basically are like this saying using machine learning learning algorithms and and looking beyond just your credit score to help people with no credit score or people who are rebuilding their score get to a better place one of the products that we have at lambda that tries to help with this mission is our credit card offering we have two credit cards right now that we offer to our to our customers and one of the big pieces of the overall ecosystem of the credit card systems is running data pipelines that move data from external providers banks and different kind of heterogeneous mixes into our data so that we can offer that data to our users and that's kind of what what motivated the the Bill of this library Bluff sir that basically had two main goals one it's to provide a common algebra 2 model network file operations and and we're talking about downloading COBOL files from SFTP servers in some Bank to storing data in s3 to moving data into port databases so this store this pipeline is very diverse and heterogeneous so we really wanted to have a common a common interface that we could access that could offer us the same interface to all these different disparate things but also would also allow us to test our software a lot easier regardless of the storm permutation that you have on the back if if you have a kind of a large algebra you can provide a pass interpreter for that algebra that will present some some scenario that you want to test them and for the second goal I mean we are a growing company and we need to scale and we need to be able to support a lot more users than what we have right now so we are piggybacking on FS to readme file where we say we definitely want compositionality expressed in a resource XT and speed definitely if FS 2 is the right tool for streaming in this case and it was really helpful on taking us and providing those tools that we needed so that we can compose the streams in a way that makes sense and and we're gonna have some samples of that in the next few slides but but basically this composition of streams with constants usually use of memory has been key to allow us to build a pipeline that we know and we feel confident that we'll be able to scale to as many users as we need to ok so this is a store and there is a term that describes this very very clearly anyone who wants to take a guess about that but i but i implicitly i explicitly live left that term out that the term is like finally tactless so this is a finally tactless algebra and the reason why i'm not putting those terms in in the in the presentation if you say don't feel like I'm an expert in this kind of category theory concepts even the the algebra an interpretive counselor is something that are new to me but but this project was definitely a great learning experience for me to to work in this contest bill on top of this concept and I think that without being a an expert in category theory you can produce very good code where you sugarcoat and and and it's it's great to see the response that people is giving us on this library that that we built like in just a couple of months of work so basically a blobster is any store that can support these six operations and as you can see this is all stream based even listing a path in some directory in s3 you could or any of our stores you could have thousands of opted items coming out so you want to be streaming that data out of your storage mechanism you don't want to have this whole list of paths in memory getting a path is basically just a stream of bytes when you give when you give us a path to a file what you're going to get back is a stream of bytes that you that you can then run through any fs2 pipes to convert that binary code into text or whatever you are converting into the food method is just going to reach it excuse me it's very return a sink so any standard FS to stream stream of bytes you can just pipe it into the sink and then whatever streams whatever buys you you use pipe into it it's gonna restore in in that path in your store and then operations like move copying and remove these are just unit operations that are wrapped on your Mona and we could have implemented like I generally move because it would be just doing again and doing a PUD but then we want to let the stores optimize for the use case for example in s3 you can use the s3 move operation internally inside of s3 so you don't have to transfer the vise down and then up again so that's kind of why we added this methods even though you could well accept move and copy you could potentially build them based on get and put we still want to let the stores optimize for for those use cases so a basically like we said anything that any any store implementation that provides all this art six of operations can be a blobster and then how do we guarantee that this stores are consistent and they have the same behavior that you can swap them in and out regardless of the the interpreter you're inducing underneath it's because we provide the library provide an integration test suite that you can implement on any storing time on any store in implementation if you if you extend that test suite then you will get a comprehensive list of test cases that will execute each one of this the base operations and they will make sure that your tests if your implement in your own store that your store is it's consistent with everything else that that we've built so that we can replace it with any other store as needed [Music] ok but oh and also one more thing I wanted to talk about the store it's also something that that the library provides is that we have an implicit syntax that you can that you can import into your scope and we will provide like additional methods like a get that uses the default chunk size of for case that is what most people use and then you can have a that receives not just the path but a file and the path and that will give you an F of unit we implemented some of those most common operations one that we use a lot is transfer so if you say SFTP store transferred to my s3 store and you give the path that the source and destination path then we will use the gate and put method to pull from one store and write into another so that's a those are convenient semantics that the library is also providing you just by importing the implicit syntax that that we are that we implemented so let's look a little bit into this or implementations and I wanted to start with a very simple straightforward case which is the get method most of the stores implementation use some SDK in the back for the file so we're just using like if you want to read a file from the file system you just use FS to i/o and read all from the file I have an implicit conversion from path from the blobstore path to a neo path so I don't need to like do the conversion every time there's it's implicit in the library or it's implicit for file store I should say there's also they get for the s3 store is just using the get object contact from the s3 SDK and then that's just gonna give us an an input stream that we are just going to run through the stander fs2 al-rayyan put stream and the same goes for the SFTP store so we we can see that there are some operations are very straightforward to implement in the store and and most of this network file storage systems have some way of giving you an input stream of the contents of the file that you want to download so to look at a little bit more interesting case this is the s3 store list function where so we start for those of you that are not familiar with there's three lists API works you basically have to call this list object function with the new with a list of the requests that will give you an object listing instance that instance may be truncated if it's if you're listing more than a thousand items so they they automatically paginate that response so you have to ask if if it's truncated then go and fetch the next one but the function you use is a different function is not like calling the same function all over an hour and again so the way we implemented this is like we went on and use the FS to unfold segment eval function so we're basically unfolding a function and you can see kind of this the signature of this initial state is a function of unit to an option of the list objects that we want to extract from s3 then in the first iteration we're going to call that function get object listing is going to give us the initial object listing and then and I should start okay there you go in the next iteration I definitely want to ask if it's truncated and if it's truncated then we can call the the next package of objects and at some point the the response is not gonna be child carry anymore so we respond none and then we know that we can terminate our stream so this is one of those that definitely took a little bit more more time to implement and it's not as straightforward as it gets that we saw before but but it's really interesting how using the regular FS to stream functionality we can we can build a really streaming solution for this list function and then a really really interesting one it's though is one that uses blocking api's and this is kind of like the exact same function we have for s3 and box store box is just for voxcom we use that to transfer data to our business stakeholders so when we call upload file we have a blocking call that is just expecting an input stream and yeah sorry that it's not very clear with these names but basically what you pass in is any excuse me an input stream and the SDK will just block your program while it reads from the input stream and whenever that input stream is closed then it will release and then your father has been uploaded so the way that I found to or the solution that I found for this is using the plain Java and all like five input streams we have we have like the input stream is just five into an output stream that then we will write with the with write all all the bytes to this output stream function and now we have two streams that we need to run concurrently one that is reading the input stream the other one is running your bytes to the output stream and then we use the concurrently function from FS to to just run both when right bytes is done it will close it will close the input stream and then it will close the output stream and then the whole stream is going to terminate and your file is going to be uploaded to the box the box website notice also that we are going back to our goals we want to resource safety right so we are using the FS to bracket function to initial are sort of pipes consume them and if something happened and we never reached this close method then we will still release those input stream in the release function in the end so we know that this is a I forgot about my whole little blobs but yeah okay so we we write and read and then we safely release our resources in the end so again going back to the store goals this these implementations are all doing constant memory they're being breezers function and they definitely first so one more you see is one that is not pretty it's not it's not the prettiest the one that I had to write and like get it done because I I still have like work to do and gets and meet deadlines so what happened here is that I use the default for the SFTP store we're using JSC H as our SFTP client from from Apache so I use the default channel list for the list operation which uses a default entry selector this is just an interface that they asked you to provide when you're listing items and this thing matches all your responses in memory so this implementation is definitely not good a good at streaming implementation because again we have this batch of items in memory we are living with this because we know that the SFTP service that we're talking to do not provide us with that many files it's just like a few files a day so it's definitely less than the thousand that that s3 uses as its default number but yeah this is definitely not cool and we know this this is a work in progress there's a lot of stuff that we want to improve but there is a path to solve is very easily which is using instead of using the default entry selector that they provide we can implement our own intro selector that uses in FS to queue and then on every item you just NQ that item and then our stream are this the stream that we're going to return is just DQ in that that just calling DQ on the on the FS to queue that you're gonna be using for that so definitely asking for pr's more things that we would like to see implemented is like an HDFS store we don't have that we haven't had the need to implement that but I don't think there would be it would be very hard to do so if one of you guys is using HDFS a lot and when I consider that it it would be great addition I don't know just like we have boxes calm we can have Dropbox or one that I would really like to have to facilitate our our mocking of these stores in a unit test would be an in-memory an in-memory store that that doesn't need to have like a file system or anything like that behind it it's just saving this this blobs of bytes in memory that would be very convenient for unit testing so use cases and how do we use this in our pipeline and really this is one of the things that I'm really really excited I mean the fact that we were able to implement this when I thought about it I say well in theory it should work this should be fine I should be able to come to compose these two streams and and and it should be good so the fact that I was able to do it and it worked is it's very exciting so we have a very standard stream of data base insert and here we're composing with do be to insert in our process database so we use the standard store get from whatever start we're reading from what our serve was provided to this to this program we use with we we build our own a binary decoder pipe that will take the pipe the the stream of bytes and then given given a certain schema it will provide us with generic records then we have this record handler tied that is basically a function from a generic record to a connection I do so we use that and we receive that here in our in our in our stream we we use that handler to so given the the stream of generic records we translate that and thanks to do we that provides a conversion from i/o to connection i/o we can translate that to a connection iostream and then a map are streamed to the to the handler and what this is giving us it's basically that the insert strip is is our database transaction that's all the inserts we're not doing the database is giving giving this one file that we're processing but the requirement is that we keep very good control on what on the files that we're processing we do not we want a guarantee exactly once processing of this file we want to guarantee that anytime that a file gets processed if then gets marked as done in in our states in our state machine basically marking as done is just moving a file on the store from one place the open the open directory to the done directory so what we're doing here is combining a stream of database transactions with a stream of store operations and then if either one of them fails we want to roll back our transaction so thinking about doing this with any other tool you would have to like do the inserts wait hold the transaction there go do this and if it fails then you have to worry about all these conditions but in our case we just translate the store operation stream into a connection iostream because the connection here is just like any other effect monette and we we concatenate these two streams and then we transact them the whole thing and like very impressively for me was that this actually works and anything that happens when you're trying to mark your file as done will actually revert your transaction and this is like demonstrating how composable the strings are and how you can makes different different string from different types of operations but the other goal of our pipeline was also that we wanted this to be easy to test and we want this to be really easy to mark and provide make sure that you can that you can test we start having all these dependencies and all this baggage in your unit test and basically for this one function the way we test it is by providing custom store interpreter that models a very specific scenario in which you have your input data so these records that are coming through our stream we know that they can be inserted I mean it's going to be fine there's nothing to be problem with them we also provide in the same store we also provide a failure scenario on a specific command when you try to move this file from this this path to this path and going back to our move function that we have in the slide before it's matching that that scenario then we want to raise the error hey this fail and and this is how we simulate our failure scenario then we provide the store simulation into our forward flow class and then we can execute this and basically what the test is doing is that after I run my workflow with this specific store I should run a bunch of queries that validate that none of the data that was supposed to be inserted in this transaction should be inserted and I know because the inserter records in the end must be 0 so not only you can compose and do all this great thing with the FS two streams you can you can also have your custom test store that will provide the exact scenario that you want to do to to confirm that your code is working in your you know that your unit test coverage is very good so yeah I I think it was a very interesting process I'm really glad that that lens supported open source in this and happy to see the response that you guys have to to this library um this is--these all I got for today so thank you very much for coming and if you have any questions [Applause] yeah so the comment is that definitely doing the in-memory implementation should be super easy and and yes that's that's something that that we know it should we see kind of that test or that I showed it's it's it's an in-memory store but it's with a fixed set of data we can definitely we know there's definitely tools that that would allow us to do that very quickly it's just that as soon as I got the bare minimum I had to go meet my deadlines so that's why we haven't gone beyond that one of the things that we also want to improve a lot and that's that's all on me I have to get down to it it's actually running that those integration tests that I was talking about that are generic for any store implementation those are integration tests so right now running them in Travis it's it's hard it's not impossible so I have to make some time to actually set up some like test bucket that I can like put my secrets in Travis and actually execute the integration test on on my public Travis wheels if you go to our readme right now our test coverage says like 26 percent or something like that the reason is because we're not running the integration test on on the public Travis but if when I run it in my local environment setting up all the integration tests we go up to like 80s mid 80s so yeah it's definitely something we there's a lot of stuff that I would like to do to it but but yeah time is limited what is the compile time for the story itself it's not that hard it's not that long the compiler actually behaves very well with finally tactless approach there is not much that the compiler is actually doing in our case because this is just an interface and then specific implementation of that interface having said that we have a broad scheme as with almost 200 record 200 fields that is all the data is coming from the bank for a given like thing so compiling those are becoming increasingly annoying so I was very interested in in Jeff stocks before because it would be nice to see and we're using for the Avro encoding and decoding we're using the Scalla for as macros just there macros module and basically it uses shapeless under the cover and we know that as soon as you put in shapeless with more than 100 fields then compile times are growing exponentially so we would definitely love to see compile times go down a lot yes definitely the the the finally tagless concept it was described by oleg i don't recall the last name but they call the famous all like he described this a few years ago and and it's kind of like in opposition to the freeman are not in opposition to free mana the argument is that if you don't have an algebra that when you are interpreting it it's like really composed and you're just like flat mapping over your over your type classes then it's it's probably not necessary to do a free algebra to to do this kind of specifically examples and in our case it's definitely not it's either one of these type classes would be the six operations that we have so when when I was talking into this and and seeing that it's definitely not I read this from pollster Sano's blog posts and on comparing free algebra to finally tackle s and he definitely explains very well how you can build the same solution on both but how if you have a very flat algebra it probably makes sense to just use finally tactless and and that's what we went for it it's definitely a lot less boilerplate code when you're when you're doing it this way well I don't have anymore anything more [Applause]