Run Metaflow Run: Dive into Metaflow Runner API

Key Takeaways

hi everyone welcome to another edition of metaflow office hours uh we have a super exciting session uh coming up today you know just a few days ago there was this announcement about the metaflow runner API this has been a long requested feature providing a way to run metaflow flows you know programmatically uh and has use cases for like you know automating these flow runs in cicd systems executing Flows In notebooks Etc one of the main developers behind this feature madur tandan is with us today madur will talk more about what the runner API is how it works and show us examples of how to use it madur has been working in data science data infrastructure and python for several years now he was an engineer at Sundial before that he was a scientific software engineer at Quan stack for a couple of years he's been a researcher he's been a Google summer of code participant for a couple of years I believe and a speaker at conferences like Pi data madur welcome to metaflow offics over to you um well hi everyone I'm madur and let's begin I guess without wasting much time so I'm going to quickly share my screen and I hope you guys can see the blog that we published about the runner so um actually before going to the blog or perhaps the documentation let's start with examples directly so let's do that so here I have a very simple hello flow it has three steps start mid and end and it really does nothing it just prints three different statements start hello world and end right so I have some Snippets here that I can show on how you know it is supposed to run so we first import from metaflow import Runner and to the runner we pass a flow file now the the flow file is actually present in the same directory so I'm just referencing it by helloof flow. py which is present here which I showed you here and uh by default the show output flag is turned to be true for uh but yeah I'm right now turning it to be false uh the runner uh this context manager can also be used directly with the Run function so you know you can use it in one single line I mean I split it in two lines because it is very long but you can imagine this is all just one line right and you get this result object now this result object is very interesting it has some properties it has a status it has a return code it has run run is basically met flow's own run object and through which you can access you know the store data inside a flow and stuff like this and property such as do finished and stuff and then at any moment of time if you want a snapshot of let's say the standard output or the standard error you can just print result. SD out or HD error uh in this case this is a synchronous version of the API which means that we will wait for the whole flow to finish and since you wait for the whole flow to finish the result SD out is nothing but the whole SD out itself it's not the middle part of it right it's not a part of it so let me just run uh python Runner demo to Showcase how it looks so I get the status as successful I got my return code AS zero my run object is basically hello flow and SL some run ID and the snapshot of STD out is this now this was very fast because this flow has literally nothing doing so yeah it ran in couple of seconds so that is how uh the API is supposed to look like now of course if you don't want to do things in a single line uh I mean you can remove this dot run aspect of it so let me comment out this one and let me go to the other snippet so basically what you can do is you can just instantiate a runner and inside that call runner. run right so I mean what I'm trying to say here is you could either do do run directly or you could do it you know separately as here uh then I mean it's just the same snippet so yeah wouldn't matter the same output the same result everything is the same right so this is an example of a simple flow running running the asynchronous API with the context manager and multi-line this was a single line so now you can also use this API without a context manager of course so you can declare the runner object for you and then you know again do the same things that that you were doing earlier the only uh Goa is you need to do runner. cleanup why because uh we create some temporary files and uh otherwise they will be you know left hanging in your system so uh it is just a you know way to clean them up delete them so to speak so yeah um if you do this you get the same output again so I just showed you three different ways of doing the exact same thing uh one using do run directly with a context manager one with a context manager but do run in another line and then you know without using context manager at all now where it gets interesting is right where it gets interesting is that if you remember the command line um how metlow is used in command line you basically have python flown name. py and we can do minus minus help here right so that will show me some parameters right so met metadata is a parameter requir is a parameter environment data store Etc now all these parameters you can pass and you can pass them uh basically you can pass them at this level so at the runner level when you're declaring your Runner object all these different parameters you can pass here right so the flow flow level parameters and uh if you do if you were to do python flow name run minus minus help then this has its own set of parameters such as Max workers Max num splits and so on right and those those parameters you can pass to the Run function so these are the top level while declaring the runner and while actually doing the run you can pass uh these parameters here right so once again to revise there are Runner level arguments that correspond to python FL M py minus minus help examples are quiet met data environment data store Etc and these you can pass here while declaring the runner object and uh while doing the dot run uh you have some run level parameters such as Max workers ma transs whatever and you can pass them here right so we can try um oh sorry we can try running this of course and uh yeah it ran and uh it ran very fast uh right now the great thing about this API right it has sanity checks now what I mean by that so first of all uh if you remember the flow level parameters using minus minus help at the flow profile you get pilent right now pilent is a Boolean type imagine me passing three uh right to pilent so type is Boolean and I'm passing an integer instead so uh if you try to run this you will of course get an error so invalid type for pilent we are expecting an optional buol and the default is true so that is good and uh what so that this type checking on the Fly number one number second is that let's say I am you know passing a parameter which is not supported by metlow for example I'm passing here Mur equal school right so here also we have you know type checking on the Fly and and basically checking for valid arguments and if I run this now I will get unknown argument madur and possible arguments are tags Max workers and Max units and so on so you have um so you cannot pass unknown arguments and you cannot pass uh wrong types basically and now once I have corrected these two mistakes then of course I can run this demo again and yeah it will just finish in in a second because it is that quick uh because the flow really is nothing right now what I can do further is let's consider another flow now so this was a hello flow now we've graduated a bit so let's come to another flow the only difference this other flow has is I'm storing cell dox equals 3 here right that's literally it so I'm storing uh data inside it now what I can do is of course I can pass my Runner level arguments which I discussed and the Run level arguments here now metaflow has this something known as The Run object and uh result. run gives you the meta flows run object and you can query what data was stored inside your flow using do data and the variable that you stored it with so do X right so if you run this now you would get X is3 and that is correct because we stored xs3 here right so you can imagine you know storing your accuracy if you're train training a machine learning model and accessing it later and using it to compare in your cicd if you're training was correct or not and stuff like this you can even imagine something like you know you see the return value of x and then if x is greater than five then execute whatever flow a if x is less than five then execute flow B so you kind of sort of construct a sequence of flows based on these values accordingly right yeah so we have discussed how to access data stored within the Run object you discussed that arguments are sanity checks uh type checking on the Fly and you cannot Supply unknown or whatever any random parameters as well now let's come to another example of a flow and this contains a parameter a user defined parameter called Alpha right and this Alpha is how many how much seconds should I sleep right so uh it is required and the default value is one so so in the flow I'm just you know printing let me sleep for Alpha seconds and I'm waiting for that time and then my flow proceeds to to the end right now the good thing about this Runner API is it dynamically injects all user level parameters also so what it means here is that uh whenever you would the command line version you would Supply minus minus Alpha right when you would do run but here you would Supply it here uh because it is a run level argument after all right now you can notice that I'm missing an alpha here right and I and I can only Supply it as an integer because the required was true and yeah the default is one and the inferred type is integer so when I try to run this of course I get missing argument Alpha is required so the beauty about this API is that the type checking and the you know uh the validation of missing arguments and um you know unknown arguments that you supply it also works for you user supplied parameters so they are injected at the correct place and this would also work for let's say trigger commands of AR overflows or SE SE SE functions and so forth right so they are injected at the correct place now let let me Supply Alpha of course because it is asking us to so let me Supply it as a string and that would be my name and of course it should result in an error again because it is expected to be an integer and of course we get invalid type expected integer default is one so let me sleep for 5 seconds now uh so this flow should take 5 Seconds to complete and if I run this it is waiting for 5 seconds and yeah it took six seconds because 1 second for the startup time setup time right now all these were of course your running flows locally uh we can also I mean if you have you know your meta flow set up correctly then of course you can ship to kubernetes and other remote execution environments and if you remember the way we did it was uh you know Python and your flow name and your minus minus with kubernetes and so forth right so with is a reserved keyword in Python uh so instead of with we have something known as deos specs here and uh since the Run command comes after oh sorry so since the Run command comes after uh you know uh the withd with argument in the command line so that's why this WID is nothing but a flow level argument so that why that's why it comes here in the runner uh you know Runner arguments and not really the Run arguments right now of course uh this assumes if you want to ship to kuun that your meta flow should be you know um set up correctly and stuff so I have another terminal where I do have the kuber set up so let me try to run it there I think it might ask me for my authentication and stuff so yeah just let me quickly yeah it did right yeah let me just put my code one second and let me cancel this because of course the code wasn't there so they were all issues okay now since the code is there let me just run now this will actually run C kuties for you because we are supplying Deco spec kues and r r so every step will be you know decorated with The kues Decorator and Alpha is 10 so it should sleep for 10 seconds but since it is on City so it might take a bit longer to set up the pods and stuff but the flow itself should take 10 seconds right and here if you set the show output equal actually your show output is true should it show the the Pod name and the logs from the Pod oh okay it will show whatever you get an STD out normally yeah and yeah there it is nice yeah so Port is running container is running and so on so I guess we can wait or we can also skip ahead um or maybe let's just wait it's it's not long anyway uhh yeah so now it is actually sking for 10 seconds I guess because that was the first task now it's at the middle task so this Runner API code itself is running on this on this workstation that you are on right like on this machine that you current local machine it's my local ma yeah local machine and from there it is essentially running what otherwise user would have run like python FL py run with kuber with kubernetes run it is doing that but also passing in all the arguments and kind of sort of what uh streaming in the result the standard output and showing it here and so on yeah so this is something uh so we saw it worked now let's come to another flow another example of a flow now this is a flow that can fail non- deterministically so I have you know I'm cing a I'm tossing a coin basically and with 50% probability you know I'm trying to convert an a string to an integer and with 50% probability I'm doing the correct thing so obviously I can fail half the time I can you know succeed half the time so this so this particular step the mid step can actually fail randomly for me yes uh so what I can do is so this Runner API also you know has resume right so uh basically I have my Runner thing again then I have my number attemps number of attempts is one then result is run. run now uh T call that result had you know a lot of properties which I showcased in the first example if the result is actually failed let me try again let me increment my number of attempts and then do runner. resume now this resume is of course if you used resume functionality of metap you know that it resumes from the fail step uh so you don't really execute from the start so if I run this uh the the attempts would be much faster than the you know the whole flow because it's just resuming from failure right oh sorry I'm actually running running on a context which is you know which is configured for remote execution and that's why it was taking time so let me switch back to my normal terminal instead right yeah so yeah this finished much faster because that one was for cloud right uh so it was yeah it is it took only one attempt but let me just run it once more and see that you know it's it's actually non- deterministic yeah so five attempts so on an average if you run this thousand times with the probability of you know yeah you'll get your number of attempts to be half because it's probability of 50% yeah so we have run we have resume now all these examples that I showed you they were basically syn chronous API of of the runner we do have an a very cool a synchronous API as well now I'm going to show that as well here so of course with everything asynchronous I'm wrapping it inside an asynchronous function uh because python doesn't support top level of Weights by default but here what you can do is it is a it is an example of a single line again so I have the runner here and I'm also calling asynchron uh right and everything is one line I mean it is longer that's why I split it up for readability but everything is just within one line right and of course it has these properties again and again but what you will notice here is that uh in the previous examples if my Alpha was 5 seconds I was getting these results after 5 Seconds yes because it was actually waiting for those for the flow to complete here I I'll show you that I'm waiting for 20 seconds but I'll get these results instantaneously because it is an asynchronous API and what it means is that you can you know play around with this result object and as I said the STD out is a current snapshot of STD out at that moment in time and since the flow hasn't finished yet because 202 wouldn't have passed so the the snapshot would be kind of you know not not full in some sense so if I run this yeah so you get status as running your return code is none because it is not completed yet uh the Run object is this the SD snapshot is this at that moment in time and uh the process ID was 8004 and I can actually you know go in PSO grip metaflow and I can see the process ID was uh 8004 as you can see so it's still executing in the background here yes because it is an asynchronous API right and if I do want to wait for I mean let's say I want to continuously read the standard output like is there a way I could write a oh sorry sorry go sorry now of course I can wait for you know the flow to complete and the way to do it is using this await result. weight that's it and the cool thing about this the result. weight is that it returns itself so what I mean by that is you can do result equals to a weight of result. weight because it returns itself so then you can access other properties again and again but let's just wait for it right so I'm waiting for 5 seconds here and I mean here you are converting your asynchronous version back to synchronous version so you might as well use the synchronous version which I showed before uh yeah and it took six seconds as you can see here right uh now this weight function that I showed you right it has some cool things so uh I can wait but I can also Supply a timeout so I want my flow to be completed within let's say 7 Seconds now of course here it would fail because I'm passing Alpha 20 and I will wait for 20 seconds but I'm passing a time out of 7 Seconds which is lower so let's see what happens in this case so yeah so you know we had a timeout which was incurred after 7 seconds and the underlying metaflow flow was killed right and uh it was supposed to run for 20 but it killed itself after seven because of the timer right if you don't Supply anything it of course waits till the end which I showed which I already showed now this weight function can also take in a a parameter known as stream and uh you can actually stream logs if I pass STD out here we'll be streaming standard out if I pass SD error here we'll be streaming SD error so yeah so I'm waiting so this is live streaming from the synchronous run right and yeah we've already waited for 5 seconds and we are kind of done right uh now of course you can use both stream and timeout parameters so stream SD out and timeout 7 right that will mean you are also streaming while also timing out at 7 seconds and of course it will fail after 7 Seconds yep cool so now we can have more control over log streaming so let me uh uncommon this licated example that I have here so what happens is that I mean this statement is similar say before let's say I have an interesting position a variable called interesting position and that is zero for now uh we also have a function known as stream log and result. stream log you can pass it you know uh the stream the SD out or SD error or whatever and then you can you know loop loop on it basically so you can loop on it line by line right so you're looping on it and it it returns two things so position and line and position is calculated after the line has been read right so I read a line I print it and if I find Alpha is substring in that line then I record that interesting position and I break out of this Loop right now I ended streaming basically at this interesting position and I can also resume streaming from a particular position so the stream log thing actually takes a position argument as well and I am resuming from there and then I'll print lines only from that particular position and not from the beginning right so I can read logs halfway as well uh and then one and by here what would happen is that the by here what would have happen is that you know the um the process would have ended by now and since it would have ended and if I try to stream again from scratch this will be instantaneous because it has already ended I I am just reading the log file right so let me showcase um that as well so basically we'll have three outputs three chunks of standard out outputs so here is the partial one the beginning I ended streaming at 232 and I'm resuming it from 232 but since I'm waiting for 20 seconds to get over because it is part of the flow that's why you don't see anything here yet right yeah and then once you resume streaming you get the logs after after that position and then once you're done uh you basically stream again from scratch but this was printed instantaneously and you can verify that because the total time of execution is still 21 seconds because my flow ran for 20 seconds right so let me come in this how does well so this this offers you a more fine grain control so to speak over your log streaming where you can iteratively you know search for substrings in your logs and so forth let's do something even more cooler so here I have a function called emit logs it is basically a wrapper over you know result. stream log it just Loops over it and whatever a supply a I can supply a function let's say a custom logger in this case it is a print function but imagine you are shipping logs to you know U let's say another service or a SAS tool that you use or whatever right and you can use that particular logger here uh so it is basically iterating over the stream log and you know using custom logger over that now I create my Runner object with the particular flow right and I have two results here result a and result B I run basically two invocations of that flow but with different parameters one with alpha s 20 and one with alpha s 30 and then I create two tasks and two tasks are nothing but invocations of emit logs I pass the result object I pass which stream I want and in my custom logger I modify my print function to have these prefixes stream a and stream B right and then I'll wait for both of them to happen in parallel and together right so if I run this so so you see stream a logs and stream B logs so task is starting for both of them right and then alp2 for one Alp 30 for another yeah so you can do cool things like this with the asynchronous version and of course since they both run in parallel so the total time of execution will still be 30 seconds because the 202 one is subsumed under that running time because both are happening in parallel right yeah so 32 seconds as we can see the execution time and both logs appear here and that turn in parallel right [Music] um then uh of course I can launch many parallel runs right so I have my Runner here I create a an array of runs that I'm launching each waiting for a different amount of time period and uh as in when I get the result I I'll just be collecting them and showcasing that hey this run completed that run completed and so on uh so I'm launching five runs here uh one sleeps for 5 seconds the other for 10 and so on and as you can see all five are running then one is completed then four are running then another will be completed and eventually all will be completed so this asynchronous API gives you uh you know ability to stream logs capture the standard output at a particular snapshot and so on now for the last part of the demo after this I'm going to show you how to um you know run this inside notebooks so I'm switching here to my notebook so for Notebook as well uh the thing is we have a different runner for notebooks because we don't have a file uh so it's yeah you basically import NB Runner and it's of course built on top of the runner and imagine um imagine imagine your flow you know having being WR in this cell only so the thing about there is a small Goa with NB Runner uh it needs to be initialized in the same cell in which you have your flow so basically since your flow is present here in this particular cell so it has to be initialized here so this is one small Gaza and U of course then it mimics the runner because you can set your you know the flow level arguments here and when you access the dot run function of it then you have your run level arguments here so it will come with all the sweet type checking and validation of arguments and stuff uh right but yeah you can but yeah you need to basically have it in the same cell the inst instantiation instantiation of it so it needs to be self-contained in that sense so once of course you import it and uh this is also non-deterministic it fails with an 80% probability so yeah we just trying we'll just wait for it to you know run it's basically the same I I can actually just chop it to save time but uh it's it's basically the same exactly the same usage right so runner. run then TR is one when it is not successful try again runner. resume yeah I stopped it yeah so we can ignore that and of course you can print the SD out SD error status Run run. finish where the loog files are the return code and so on so this is your NB Runner the synchronous version of it and since I showed you the asynchronous API as well we have the NB Runner asynchronous run right so when you have asynchronous run you get the result immediately and you can do result. status which is running of course and I showed you the weight API so you're waiting for the log streaming the streaming of standard output and then since it failed because it was failing with non- deterministic probability I mean you know 80% probability and then and of course if you print status it has failed and just like we had an asynchronous run we also have an asynchronous resume so it is uh very similar it's just the same thing essentially right so r. async resume and yeah if you do the status you have running again and if you stream it out and this time we got lucky because it actually passed and here you have result. status which is successful so this is these are two examples of you know using NB Runner with the synchronous run synchronous resume and the asyn and asynchronous with log screaming and all these things I have one last example which makes things even simpler if you don't want to bother with all this just use NBR run NBR run is a very short you know short cacha for hey I just want to run this give me the output that's it nothing nothing more to do uh and yeah it will give you the output and similarly you have NB resume so if you yeah if you don't want to get into the you know best practices of using it you just want a quick of doing this just just use just use NB run and NB resume so yeah this is just for you know making life more simpler but otherwise if you want the full feature set you should actually use async run uh run and so on so yeah this wraps up the you know demonstration of the runner API and uh to go very quickly into some uh you know lower level details it's three layered architecture uh we have a lower level API we have a subprocess manager and the this higher level API is built on top of it and the lower level API is where the magic happens of type checking and validation of arguments and so on right so that is something uh and it's a bit of reverse engineering uh because this is the this is basically you know if you execute python flow name. py then you have some arguments at this level and you have command groups run you have command groups or kues and kubes has its own command step so c c c denotes a terminal command here g g denotes a group right and Argo close is also a group and then it has several commands as create and Trigger right and at every level you have these arguments or parameters so we basically it is a tree data structure we basically reverse engineer and pass this tree structure at runtime to construct the lower level API and uh as you can see the user defined parameters were Alpha right so Alpha and my file for example they are injected here D ically for the Run command and wherever it makes sense so it also makes sense in trigger so that's why they are injected here as well so yeah that is some interesting reverse engineering aspects of it right uh yes and of course if you want more information then we have a documentation present here which uh we can send on slack and we recently also had a blog post about it uh for upcoming there might be some bug but happy to have feedback in fixim and uh um basically we also have just to just to give a teaser we also have upcoming runners or let's say deployers for ago workflows and step functions which will have programmatic apas of doing those two as well so yeah hope that that's all