Scale By The Bay 2020: Yves Parès, Reproducible Data Pipelines Using Controlled Effects

Scale By The Bay 2020: Yves Parès, Reproducible Data Pipelines Using Controlled Effects

Recording: Scale By The Bay 2020: Yves Parès, Reproducible Data Pipelines Using Controlled Effects

[Music] hi hello everybody hello uh so thank you all for uh coming to my talk today so i'm gonna talk to you a bit about what i and other people at twig have been doing for more than a year uh in the context of task and data workflows and like to bring to you today the approach that we came at which is a general framework for uh making composable tasks and data workflows so exposing tools to both engineers software engineers data scientists and data engineers uh in order to implement workflows and to uh to provide them in a robust and reproducible way so where do we come from twig is a company it's a french company that's been existing for about eight years now we do consultancy on functional programming projects uh various aspects of it develop general development infrastructure as code data engineering projects we have no preferred project scope we've been having projects ranging from a few weeks to a few years my last project for instance has been going on for three years and no specific domain we can go work on projects ranging from bioinformatics to finance to the financial sector and as for me well as like i said i'm a haskell developer and a data engineer and i've been working mostly i've been working mostly uh during my phd and then after a tweak on bioinformatics projects so i'd like to start this presentation by giving you um i think it's just a quick categorization of uh what we mean by workflow tools because this is something that might be a little um obscure or a bit fuzzy when uh we don't know exactly the terms because some are not exactly super well defined so i don't have the pretension of defining any uh terms right now i just want to be clear at what i will be told i just want to make clear what i will be talking about so first i would like to make the distinction between what i would call a task workflow and a data pipeline to me a task workflow is a direct basically graph of tasks and the point of them is that they are eagerly executed so one task completely terminates before the next one starts so it's in a way similar to the way build systems work in order to build a target you have some intermediary targets and they all need to be completely evaluated completely computed before we can go to the next target and we have some level of parallelism which is possible but only when two tasks are independent meaning when they're not directly connected in a graph so tools we have like that's in the domain of data science would be luigi and airflow for instance and what i mean by a data pipeline is something more similar to apache beam of storm it's also a dag also a graph of but this time of data processors and it's more similar for the people knowing uh lazy evaluation and functional programming it's more similar to lazy function processing lists and in that case all processors may run in parallel because when one processor yields new data then it can continue to output new data when the downstream processor is processing the data it just output the second difference i like to make is between tools targeting fine grain tasks so where each node is just one elementary operation for instance in terms of flow each node is just can be as simple as one mathematical operation and workflow tools that target cause grain tasks where each node is a script or a docker container running a script like again widget airflow view the third way to categorize workflow tools is more noun it's the distinction between static graphs which where the whole set of tasks has to end their links has to be known in advance and if you have something like conditional branches for instance if this task yields uh some number then go to that task or else go to that other task then all the possible branches maybe must be known also in advance and that was the case in terms of flow before version two and this is i believe also the case for luigi and airflow and the other uh the other class would be uh workflow tools working with dynamic graphs where uh the computing as the processing goes it may add new nodes uh to the compute graph and this is for instance do a pi torch or apache spark work and the final categorization which i haven't found much mention of is between workflows that only target what i would call the process time which is when the actual logic of your application runs and what i would call the configuration time so the distinction i make between both both happen at the runtime of your program so everything has been compiled or everything has been interpreted but at config time you may perform some configuration like for instance choosing which data sources and syncs you will target choosing which http endpoints which databases you may perform so ahead of time ahead of time checks this way so checking network is reachable if all the resources are in place you may perform some graphic writing in order to do optimizations and if you have a scheduler that works through static scheduling it will operate that way and of course this possibility to exploit your graph at config time requires you to have some prior knowledge not only about the graph but also about the computation so it require requires you to be able to deeply introspect each task of your graph so after this uh this categorization i will present the scope of the work we've been doing so our contribution is some approach which we call commental which we've been um for which we've published a paper at the haskell symposium a bit earlier this year this year so it's an approach that's usable to implement both tasks uh workflows and data workflows at any level of granularity and which can export also prior knowledge but we are although restricted to static graphs only and the application domain to which we've uh been applying this this technique is a simulation and data analytics applications for instance in pharmacology with pfizer or in biochemistry with the client i'm working with currently who's a which is a small french company called nova discovery and deals in simulating biochemical reactions in the human body so i'd like to throw you some buzzwords about what our tool is how a tool is doing stuff like it's making stuff reproducible like in in the terms that in the sense that you can run twice the same pipeline and get twice the same results um it's driven by some strong focus to homogeneity which means that a task and a combination of tasks and a full workflow which is just a big combination of tasks are all the same type you've got one type uh that identifies all these all these concepts so it's it favors composability too so gluing two tasks together will yield you a new task of the same type it favors configurability meaning that tasks are not tied to one specific environment so not tied to one specific uh file system to one specific database or everything so we keep the logic of the task abstract from the i o it favors reusability meaning that tasks can be imported defining modules we're used in many projects uh and it favors separation of concerns meaning that the developers world and the analyst or the data scientist world are kept separated so to to sum it up we strive to strive to make the tasks uh to be usable just like plain old function and how did we do that or by combining two different um domains of pure functional programming named abstract effects and arrows so first talking about effects effects are something that give you control over what the tasks can do so here i have for instance three examples of tasks you might find in regular data science pipelines like training a model classifying some data and or simulating a system of equations which is closer to what i do with my client currently so each each one of these tasks will need to perform some effects so effect is really well it's up to the developer it's up to them the one who models the domain of expertise to define what an effect is at the at its broadest sense the word effect can be anything that runs i o and this is the way it works usually in haskell uh people who know haskell know that you have the distinction between i o code and the pure code but a code that is flagged as io can do basically anything it can write to file it can reach some web server it can write in some database it can do anything that is just purely that is just not purely computing a value so for instance when you uh detail when you define the effects that these uh tasks want to do for instance for training a model you might say that well you have one effect of downloading a data set but you can refine it further as to saying that i want to download the data set from a specific source so downloading a data set from one s3 bucket for instance might be a different effect than not downloading it from another s3 bucket so you have full control over the granularity of what your effects allow the people who use them to do you you will want to log stuff and you will want to write some results and the same would go for classifying data or simulating any kind of scientific computing so you can for instance here and notice that the logging stuff is present all the time so if these three tasks are being used in the same pipeline well idead we would like to configure the logging feature the logging effect just once and for all we would like the three tasks to be sharing the same login system to be sharing the same effects and same goes for downloading data sets for instance you would want if the same pipeline trains the model and then classifies data you would want the two tasks to uh use the same kind of effect when they download data sets it would be configured the same way and it would be it would allow the same permissions it would provide the same features to the to the data scientists and the benefits of having so finely controlled effects are well can we move with a huge class of problems for instance when pipeline that can do a park arbitrary io well that can't exist anymore a pipeline that's sort of merged some execution logic which should not be strictly speaking related to the logic of the pipeline when it's gone too task writing to file system they shouldn't be it's gone too so it's it's sort of pushing further the advantages that we already have with static typing is that more errors that's before were happening during the execution become compilation errors and even for the errors which we can't really express as types because the way to uh make stuff become compilation errors for instance in haskell or similar languages is to encode more more information in the types even when it's complicated or cumbersome to encode this information in types you still gain um what we call early failure because your task will be inspectable at config time so remember that by config time i meant what happens before the actual logic of a pipeline runs so this gives you early failure and all of these features promotes reproducibility so reproducibility is not something you get for free but it's something that can be facilitated by the tools you use so let's delve into some code now what do we mean by abstract effects so here i call them binary we'll see later this binary term so what i mean by an effect is something that has an api and several possible implementations so it's super similar in this uh in this in the grand scheme of things is super similar to an interface or an abstract class and we the api of each effect will be expressed as a haskell data structure with two type parameters and this is why we call them binary because each effect has two type parameters and in the jargon the implementation of one implementation for an effect is called a handler or an interpretation function so here in my block of code i have two simple effects one that is just doing some logging so here are my two type parameters the first one refers to the input of the effect the second one refers to the output so here i have one data constructor so this is something that can be created in my code or pattern matched in my code and i'm saying because this is gad a generalized abstract data type a feature of haskell or camel and scala in which you can say well if you use that constructor you will get this type so the type of the input that is expected by my log effect and the type of the output that is returned by it is directly encoded here in the type so i'm saying just that this is my interface when you trigger a log effect well you have to feed it a text that will be written and it will give you nothing because there is no interesting result in this bracket type which we call unit in haskell is the way to represent uh just non-interesting results it's a type that just has one possible value and another effect would be file i o so same two type parameters a for input b for output and i have designed two constructors read and write so this time we have a special parameter here which is the file path of the file we want to read i'm saying file path but it can be a url or anything and this file path is not strictly speaking parts of the input of the effect for reasons we will see later the input of read is nothing and when you want to read some file it gives you a text and the input of right is just exactly the reverse so now that we have abstract effects the just abstract representations some data types that can be shown that can be serialized anything you want you might want to actually execute them and this is where the interpretation function on handlers come into play so you will have a new type which also has these two type parameters which we call core and which uh which it's actually allows us to perform some work so here people knowing haskell can recognize the i o stuff that so it means that this function can do basically whatever you want so if you have a to iob function it means this function receives an a and can compute anything it wants and return a b so it can do some work you can do any work and because it's not constrained it's actually not the type what we want to expose to our users the type we will expose to our users are these ones like the effects the abstract effects we have and once we have all this we can write our basic interpretation functions so we'll say that log of something is just transforming that effect to just the regular print string function and file i o transforms a read into the function that will actually read the the file and write into the function that will actually write the file so it's pretty straightforward let me just check the discord no questions yet okay if you happen to have questions in the meantime don't hesitate to ask them on the discord so very trivial effects indeed let's see a less trivial one so in a similar way i'm having a classify effect so classify is i want to well to classify some array of data some vector of doubles by using some models some pre-trained models so here about the same actually it's the same pattern than when i want to read files i'm just saying that i have some model name which is pre-trained it like it may correspond to some resources on the tensorflow hub or python or any other source when you can when you can where you can download some pre-trained models from and these effects will receive the data as an input and it will give you some flag some integer corresponding to some classes and output and the interpretation function is also slightly slightly straightforward so here we need to do a bit more than just one function we're using the model name first to download the model and then we're using that model to classify it so we'll just do the code stuff but here we see that we have something that might be improved like because our interpreter here does everything at process time it means every time in our pipeline the classify effect will be triggered every time we want to classify something by using some model we will redownload them all every time so what if i want to know in advance the model that will be used and for instance check whether it exists that i can give early failure in the case the model does not exist before anything runs what i want what if i want to do this download model operation at config time and what if i want to do it just once if the same model is being used several times throughout my pipeline so it it gives you the same advantage than caching but on top of that it gives you the advantage that you don't have to wait until the first classification is triggered by a pipeline to give an error if the model doesn't exist so it's it's encompasses what caching can do in that case but it can do a bit more in terms of early failure so we have effects what i'd like to show you now is the other parts of the work the one that's talking about how you chain effects together how you weave actually uh tasks into a pipeline so to keep things a bit abstract first i would just say that i have some type t which represent a task or my full workflow because i remember i mentioned before that the same type represent tasks or a combination of tasks and this task also has some input x and some output y and let's say i have three tasks task one a which takes some a in input emitted some b task one b which takes some the same type a and returns a b prime and task two which takes a pair of b and b prime and returns a c so from the types you can see you can imagine that you will want to glue this task together in the in the in the following graph so pretty straightforward and what errors do if your type t is an instance of arrow if it implements the arrow control structure is to provide you with two operators other stuff too but mainly you have these two operators one that allows you to do parallel composition so for instance this combo composes together a task 1a and task 1b into something that will return a pair of their outputs and sequential composition so if i have this new task i can just sequence it into task 2. and full workflow here is just the equivalent encode of this graph that i have here and for more big workflow it can become very cumbersome to express all the combinations by using these operators and there is a more syntax available more convenient syntax available which is called the error annotation in haskell and i will show you an example of that a bit later so we have effects that have two type parameters we have arrows which have two type parameters so maybe there is a link between them maybe there is something we can do to link both worlds this link is what we call triggering an effect so in our um in our framework triggering is just converting an effect into a task it's just taking some abstract effect we have and converting it into a type t and this way we can then trigger some effect here we just read some file and write its output into the log so we use the operator we saw before so we trigger an effect we get this result we get the effects result and then we feed that result into another effect so as a summary you start with effects when you trigger effects you get tasks by using the arrow operators you combine tasks into workflow and once you have a workflow which internally contains a lot of effects you use the interpretation functions to transform all your effects and your workflow into some core type and the call type can do io it can do anything it wants but of course the user the data scientist or the modeler or the uh the people who are not developers will only see that part they won't see that part the interpretation and the call type this is left to uh the developers to choose how to implement them so the remaining question we have is well for now i've been keeping this type t abstract so what should this type t which represents both tasks and workflow b so this is where our main proposal is so the way we represent tasks by using a new type which we call rope and rope basically works through dependency injection so here you see a rope is internally just a function that will transform a set of handlers into our core type so it would also be parameterized rope by a and b because remember every type needs to every task needs to have a parameter for input and another b parameter for output but internally this function can trigger some effects and in order to know how to interpret these effects well we just give it the list of the full collection of all the effects and see this is why it worked through dependency injection we give the rope uh to execute it we just give it the set of the interpreters that it needs so some terminology so here we what we call the mantle is just a type level collection of all the effects and we give a name to each one of the effects so effects are not only determined by that type the types that we described before for instance log of file io we also give a name so it's also a type level name so it's our symbols which are noun at compile time which means that you can have twice the same effect for instance if you want to have two logger channels or simulate yeah any kind of multiplexing you can have several times the same effect but under a different name and every time you want to trigger an effect well you have to specify the name the handler is just the collection of all the interpretation functions so you have one interpretation function per strand and the core is the target type of all the handler the one that can do all the i o and the function that triggers the effect we call strand and kiwis for instance our task echo from before so we say that we want first to target to trigger the read effects and here we gave a name to the strands of file io effects that we want we will call it file and then we trigger our log effects which will just receive the outputs of the spot effect and this is reflected in the type so here people who do not know a haskell this is a set of constraints that are applied to the type here our type is polymorphic it can work over any type of core so any type of um for a full concrete type in the end and it can work over any mantle but dismantle it has the constraint that it has to contain some file io effect under the files name and some log effect in the log name and additionally because we want to be able to use these operators for combining them this core type also has to be an i o because the property of rope is that it propagates any kind of control structure any kind of operators of gluing operators that are provided by the core if arrow if a core is an arrow then our full rope will also be an arrow and therefore we will be able to use the gluing operators uh on our uh and our tasks so what is it called a rope well it's called the rope by analogy of uh something in sports and rock climbing which is called the ken mental rope it's a rope that is made out of several strands we've woven together to make what we call what is called the mantle which are surrounding a core so in in rock climbing for instance the idea of such ropes is that they have a sturdy core and a more flexible mantle so you gain the best of both worlds you have at the same time something that is sturdy and also quite flexible and in our analogy all the effects will be the strands that get woven into the mantle so this is where the names come from the name is coming from so how could i extract information at config time this is where this bits that i mentioned before becomes important when i i told you before that when i write some effects i might want to exclude something from the input and just put that as a parameter directly of the constructor well this is because every information every bit of information that is not uh included in the inputs every every bit of information that is dynamic directly a parameter of the effect can be retrieved at config time and there are two two possible ways to do that either you do it in two passes so you write two sets of interpretation functions one that will just extract the metadata from your effect and run nothing or you can do it in one pass you can change the core so that it can both accumulate the metadata and perform computations and it turns out that during one pass is actually simpler so i will go quite fast over that one because i'm uh i'm not having a lot of time so here i have a simple task of a simple pipeline of three tasks so i want to read some file compute some stuff based on what i read on this file and then write the results in that other file so i want to be able to gather these file paths in advance i want to be able to retrieve them to collect them and then i want to be able to rebind them to remap them i want to be able to treat source.cxt as not being a definite path to some place but just being uh just as being some index some kind of identifier something i can remap after so what i want is to be able to what i want to do that is to be able to change my core into that each effect when it is interpreted will be well the same computation function but which is parameterized by a new hashmap which is the mapping so the mapping will describe for instance it can be read from a from a configuration file and it will say that the id source.txt is mapped to this actual file path same for sync.txt but at the same time while i'm providing some function while i'm interpreting my effect into some function that will actually perform the computation i want also to provide the list of ids that are used by uh my uh by my function so uh for instance if i want now to combine two uh call with pass then it's simply chaining these two functions and concatenating together these two lists but the question that uh we can now ask ourselves is can this type call with past still be an error because it's quite a different type from before it still contains some idea of this computation but it has some extra metadata and it turns out that yes for reasons that involve a slight haskell black magic which i will be glossing over right now because i'm lacking a bit of time but that allows me to just jump to the example i had for you so this is an example that is publicly available on github we use here uh pre-trained pytorch models to classify pictures so here is the workflow as some data scientists might work it's a very simple it's just to showcase what you can do with that so first we want to read some so we're using their annotation here so instead of using the uh explicitly the gluing operators we can read stuff in a more imperative way but that's amounts to exactly the same so we first we have two effects which one get picture effect which just reads some picture from some place which is very similar to the read file we we had before and we have a predict effect so first we will want to read some uh data some input data in the form of a jpeg image from uh from somewhere note that it doesn't we don't specify where it should be read from this will completely be up to the uh handler and then we want to run some prediction on this image we want to classify it using some model so we have this model name inception v3 and we trigger the prediction effects by saying that this model should be retrieved from pytorch herb under some collection of models named vision at some some version and then we want to predict using this uh random classification based on the exact same image but using another model which this time doesn't exist so what we want to have what the behavior we want to have with that final pipeline is to have early failure so before we even start the pipeline before we even start reading anything we must know that this model does not exist and that it will fail so internally the example i will uh leave people who are interested to have a look at it but internally it's called some python scripts using scipy and pytorch and uh the docker images that are needed are built at config time and the needed models that download it at config time so if any error should happen at the building at the time of building the docker image or at downloading an image or downloading sorry a model it will happen at config time so here i run my application so i made the log so that it separates clearly what happened at config time so first so this is this is the code exactly so this is the code that runs i have just this pipeline and the rest is just interpreting the pipeline and determining how the effects get interpreted so the code that you get is really this one so it doesn't seem from reading that that uh things that happen here will actually be reachable before things that happen here but actually all this metadata that is before this kind of arrow every everything that is here in the pipeline is actually reachable before anything starts to run so you see that first it starts by building some uh docker image so it wants to use python it wants to install scipy on top of it so i have everything on cache so it's fast and then when it's successful it notices that it will need this model so it downloads it it finds it and then once everything is uh downloaded and validated it begins the process time so it reads my image and then it writes some classification and if i now uncomment this rubbish model after rebuilding well you can sort of guess what it will do because i teased that enough during the rest of the presentation i suppose but if we just want to wait a bit stack is a bit slow i'm on battery so that's my it might be okay so it's starting and then it's saying that it will need the same model done before but then this additional rubbish model it downloads the first it tries to download the second but then it fails and it never reaches the beginning process time line so everything has been happening happening before the logic actually starts so quickly a conclusion for that so some features i did not mention first we can do what we call reinterpretation of effects which means we have effect handlers that may themselves trigger other effects so effects can be abstracted also so you don't have to you can share logic between several effects you can interpret some effects on top of an existing pipeline that sort of stuff because the app api is completely separated from the implementation then you have different environments that can just kind of correspond to different set of handlers so for instance you could have one way to interpret your effects in tests uh like targeting a simple files file system and another way to interpret them in production targeting actual databases we can provide caching so if any task if the inputs of some task are hashable and its output is serializable you can just wrap that task in a way that's um backsided by a local caching it integrates quite nicely with common haskell existing libs and abstraction and we can use other structures than arrow for instance if you want to do if we want to be able to do a conditional branching or feedback loops or that sort of stuff so the future work that we the work that we're doing right now is first heading for the first v1 of cadmental also releasing fenflo 2 which is a higher level api based on kenmental which is strictly speaking for task workflows so we have some pretty fine effects we have a an easier to use um an api which doesn't require you to deal with too much with effects and that sort of stuff we need to do some benchmarking because if the overhead is expected to be slow to be slight uh sorry because we only have some simple types like functions and tuples we still need to do a bit more benchmarkings i'm also working currently on how to distribute tasks using a similar hashing scheme than what we have for the caching and we want to experiment on more flexible frameworks than arrows so thanks everyone uh my colleagues uh matthias andoran will be available at the booth uh they will be demoing for instance some funflow two and talking about that if you're interested and i will be uh in the um in the spatial room to answer your question thanks thanks everyone you