Cloud Native Computing Foundation KubeCon + CloudNativeCon North America 2021, 30 Oct 2021

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Scaling Machine Learning Workflows to Big Data with Fugue - Kevin Kho, Prefect & Han Wang, Lyft

Description

Don’t miss out! Join us at our next event: KubeCon + CloudNativeCon Europe 2022 in Valencia, Spain from May 17-20. Learn more at https://kubecon.io The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Scaling Machine Learning Workflows to Big Data with Fugue - Kevin Kho, Prefect & Han Wang, Lyft

Data scientists often use Pandas for data that fits on a single machine, and Spark or Dask for larger datasets that need distributed computing power. What happens though, when the data starts small and then grows too much for Pandas to handle? Data scientists often find themselves reimplementing the same code to transition to Spark. Even code with the same business logic needs two separate implementations. Fugue is an open-source abstraction layer that solves this. In this talk, he'll show how Fugue lets users port native Python code to Spark or Dask with minimal code changes. By using Fugue, data science code will be written in a framework-agnostic and scale-agnostic manner that allows it to be ported to different execution environments. This will be demonstrated by showing how to scale data compute from a single machine to a Spark cluster set-up on Kubernetes.

A

Hi everyone. Thank you um it's I know we're at an infrastructure conference. So it's nice to see fellow machine learning, practitioners or people interested in machine learning, um so yeah welcome to scaling machine, learn, machine, workflow, machine learning, workflows with big data to big data with view.

A

So we're going to start off with a demo presented by han in this demo, we're going to start out with a small panda size data frame and then we're going to apply some business logic to that and trans and and bring that into spark and in the process we're going to see how much of code is needed to bring it to spark how much pandas and spark are different from each other and how the fugue opens how the open source library field can help you do this.

B

B

B

Hello: everyone, my name, is han wong and today I'm going to do a quick demo of some very interesting features of fugue and I'm going to switch between python code spark code and figure code. It's okay! If you are not familiar with with spark and if you cannot follow on the spark part, because uh that is just used to compare with fugue- um and I think the most important thing is just to focus on how you solve this problem, how intuitive your own expression should be when you want to solve this problem?

B

Okay, now, let's start! The first example is. um We have a very interesting business logic which I guess, if you're working a company, you will always have some tricky logic where, in this logic, we have, we are going to process the data frame, and here is the business logic. We have four columns a b c d. They are all strings and for any given row.

B

If the number of non-null values is less than two, then we drop the row and then otherwise we concatenate the non-null columns in order and output, the value as e-column and also keep the original columns right. This is not so difficult and the most intuitive way to express this logic, I think, is in this way. You should take an iterable of lift as a list of values and then yield those values when it meets the criteria right. This is just python code and also this code is very testable.

B

So now, let's just write, as you can see, the last two rows will be dropped because there are too many nuns okay. So this is the logic and now, let's build some um sample data for us to work on, and you don't have to care about this function, just look at this data frame, so this data frame has four columns a b c d, and it has some nouns right so now the question is very simple: how can you apply this function onto this sample data?

B

This is dependence, data frame, you know, even you know, even you use apply. You cannot directly use apply on this concat on this function. You have to do some sort of transformation. I don't have. I don't put a solution of pandas here, because we don't have much time, uh but uh now, let's directly jump in house figures, solving this problem, so remember that the semantic is. I want to apply that function to that data frame right. It's that simple.

B

Let's see how solving that so in field, we just import transform. We want to apply this concat function onto this pandas data frame and also specify output schema is the existing schema plus a e column. So that's how field is solving this pandas frame? A data frame problem? Okay, now think about that.

B

But in your real production, this data frame can be a spark data frame because you're, probably processing terabytes of data right. So when you want to scale to spark, the first thing you need to do is just initialize. A spark section. So here is for simplicity reason I just start a local smart session, but you always have your own way to get this back session right. Okay, so now the question is: how can you apply this function to this data frame using spark? So look at how spark is doing this? There are several things.

B

First, you have to bring in this pd sample into spark right and the second. The map partitions has to be used on rdd, so you have to convert it to rdd and map right and in the map partitions. It can only take a uterus iterable of rows. This row is a spark specific class, a tuple, and then you have to do this conversion by yourself, so that conquet can accept this input. Type right and another thing is after you do all this kind of conversion.

B

You have to convert back the rdd from from from the rdd to the spark data from and now here this is probably the simplest way. Oh sorry, the simplest way to um it's, probably the simplest way to describe your schema and I guess a lot of people if using spark. You are using a more tedious way to describe this, especially when the schema is complicated, and now you see we solve this problem right, it's good and it's over 10 lines of code, additional boilerplate code, but think about that. This is just one function.

B

What if you have a lot of such transformation logics? How much convert boilerplate code you have to write? Now, let's see how field is solving this, as you can see in field again, it's only the the transform function and now, instead of using pde sample, we just use spark sample, and this is not changed. We still use concat schema is not changed, and here we just tell the system: okay use spark session to run this transformation.

B

This means uh I want fugu to trans, to translate this semantic to spark operations and to run on smack.

B

So it's just one line of code and it can solve that problem, and the next example is um what, if I don't even want to convert the original data frame to spark data frame yeah? Actually you can directly say I want to transform this pandas data frame using this native smart function native python function, but I want everything happen on spark. You can also do this so this transformation, this conversion data conversion- is automatically done on the field level.

B

Okay, so now, let's think about another case, what if we have a, we have a business requirement, change. uh Remember that we had a b c d, four columns now, for some reason, the column d no longer exists. We don't want column d right so now we create another data sample called pd sample two and pd sample two. What if other business logic has not changed? Just just that we we miss one column. Will that cause any problems? Actually, if you try the exactly same thing on spark, you will get an exception. Why?

B

Because the spark logic look at this, you have hard-coded too many things. It's not flexible enough. It cannot.

B

You cannot. It cannot adapt to this, this changing environment right. So what you have to do just for this small change. You have to rewrite your wrapper and you have to rewrite this output schema all right. So it's so in real production. You will think. Oh what should I do? Should I just? Is this worth it or not? Right, it's it's! It's a struggle to to to make the technical decision right, but what, if we want to apply this fugue? So look at this?

B

There is no change at all, so you still can apply this nsc frame with uh without d column, and you can still use exactly the same function, transform function and everything else is the same. This is because in field, for example, the schema, this kind of expression is very dynamic already and it can be determined. It can be determined by the input data frame and also for the concat, because we have the most intuitive expression. So it's it's actually column agnostic.

B

You can have more columns, you can have less columns if other things don't change your functions, don't change and few code, don't change.

B

Okay, so now, let's switch to the second example, we want to do machine learning now.

B

um So in the second example we have, we have a data frame with many uh categories and for each category it's uh the the features abcde and the value the target, the y they have different. They have a different relations, so meaning that, ideally, you should train those models separately for each category because they are different machine learning problems.

B

But here now, first of all, let's just generate that training set and let's take a look. So this is the training that training data have like 400 and test data will have several thousand and we have the features.

B

um The ground truth and also the category and the if it is trained so for trinity, is all true right. So, okay, so now, first of all, let's try to train a logistic regression on the entire data set on the entire training data set.

B

Okay. So we train this data set uh on the entire data set and also we can write. A very simple predictive function predict function on this pendant's data frame using this logistic regression model and we can get the prediction right so so here is just very simple python functions. It has nothing to do with spark or anything. Fancy.

B

Okay, now think about that. How can we if, if the test data is huge, how can we apply? This predict function to that huge spark data frame? Okay. So if you want to do that in spark now you can see that the nasty uh things about spark schemas, you have to import a lot of like weird weird classes like extractor type, struct field, integer type. So look at this code. What do you do?

B

First of all, you have to bring in this data frame into spark, and then you have to you have to construct the schema so in field, it's like a star comma, the new schema, but here you have to do things in this way, and another thing is when you want to when you want to use this predict. You cannot directly use predict because in spark in pandas udf after 3.2, sorry so after after spark 3, you have to have this type of function to work with pandas udf.

B

It has to take a iterator of pandas dataframe as input, and you have to return the iterator of pendant's data frame. So so, as you can see that we have to write a wrapper for this just for this purpose. For just for this purpose, um and and and and in the end you just we- you just try to use a select statement to compute the precision where this is the sum of two positives, and this is the sum of all positive predictions.

B

B

As you can see, it works right, but this code is really tedious and it has a lot of spark dependencies right. You with this code, see how many things you have to import from spark. How many, how many like a special data types you have to learn if you want to use mark right? Yes, but what about fugue so again in fugue? It's just two lines. The first line is still transformed and transform. We can directly apply on that test. Data set no matter. It is pandas, dataframe, spark data frame, dash data frame or qdf.

B

Any data frame, netflix supports, and then the predict function is just that original function right and schema, and you can also specify the parameters to passing the the logistic regression and then spec specify this back section and this the output of this is native spark data frame. So you can do exactly the same thing here to compute to compute the precision, so we reduce this whole lot of code to just one line of code.

B

Okay, and we get the same result.

B

Okay, then we want to do something. That's more interesting, so think about that. um As I said for different categories, it's actually a different machine learning problem. What if for different categories, we just train a model and then we can apply that specific model to the uh data frame belongs to that category. So this is how we do that and first of all, let's think about python code. So, first of all, given this data frame is a category of training data right. What do we do?

B

It's it's straightforward right, so we just return the category name, which is the first value of this category column, because this diaphragm all have the same category and then the model. What is model which is pickle? We just depict this logistic regression model that we train on this category of training data right. So we return a string and a binary block.

B

Okay for predict category a predicted category. When we get a model, we get a model which is the output of this step.

B

We know what is inside right and also this df is the df we want to predict on so it's so here we just add an assertion to make sure we don't break anything fig, doesn't break anything, and then here we just the pickle loads. This this model just recover this model from this data structure, and then we just predict on this data frame and assign to the prod column right. So it's just a simple python. It has nothing to do with anything else.

B

Okay, now, let's see how we use fugue to orchestrate these two simple python functions, so it's no longer transform and actually to be honest, transform is just one tiny feature of fugue. It covers a lot more things than just transform, so here we use some more advanced features of you. So actually figure is all about direct directed acyclic graph, so we build a field workflow and we just get the train and we predict we partition by category and then we transform using this train uh cat right. So this is first step.

B

So this step we distributed train many many models and the second step we just zip this model with this test data set and then transform when we transform. We just take in this simple function which will take two data frames and then we'll operate on that. So this is a code transform in fugue and then in the end, this is a similar expression like what we did in spark, but this expression is thermal cognitive. It can work on pendants. It can work on that kind of workout spot can work on any execution engine.

B

So, as you can see after we train and predict per category, the overall precision is much higher than we train a single model on the entire data set, it's almost 10 percent up right and now. The last thing I want to talk about is physical.

B

Physical is a different way to express your logic, but it's it. The sequel will be translated to these operations. uh Well, you can keep your mindset and your semantic or in omc core. So it's it's perfect for sql heavy pipelines where you just need a tiny bit help from python.

B

So, for example, in this way we just select. We just construct the the train date, data set from the original data, the data containing both and then we just train model. So this is is exactly what we did here and then you zip the two data frames and then transform do the prediction and then, in the end you just return the result.

B

So here the only difference is that I, in the end I want when I predict, I predict entire data set, containing the training set and test set, and so here you see here, I just group- by trend group by this this flag. So we can see okay, this trend, how is the model performance on the training on the training data set itself, as well as the test data set? As you can see very simple, we just get um for training data sets. We still get better results which is expected and for the test data set.

B

We get exactly the same results as what we have okay. So this is the end of the demo. So I will let kevin to talk about some more details. Thank you.

A

Hey so what we saw from that demo is that when people move from small small data panda size data to spark data frames, even for it to implement the same logic, there's a lot of boilerplate code that has to be added or a lot of times, code has to be rewritten, but there's also other there's also other problems that happen from when transitioning, from smaller data to big data and we'll explore these in this section.

A

So the first is reusability of code in the top. The top code, snippet is how pandas would implement getting the median of each group and the second one is spark, and obviously in spark there is. There are added parameters around the tolerance, because in a distributed system, getting the median is a very expensive operation.

A

So normally you would just use an approximation instead. Second, is inconsistency so with pandas and spark, even though they both operate on data frames, they they have a lot of inconsistent edge cases. So, for example, pandas will join null with null, whereas spark will not join null records together. When you sort them, pandas will put nulls at the bottom of the column, for both ascending and descending spark will treat null values as the biggest value.

A

So it's the bottom when you're, ascending and top when you're descending, so even for these types of things, they're different, entirely different systems, and even if you wrap your pandas code and bring it to spark, then you may you may have to still uh write extra code to deal with these inconsistencies.

A

uh Second, is the third is that a lot of pandas users are not familiar with the lazy evaluation of spark. So often you find that you find that pipelines that have been moved to spark offers often suffer from uh inefficient computation.

A

So in this uh in this graph computation graph above, if you don't persist b, then it's recomputed three times for c d and e when when those are computed, so of course, if you use persist and spark, then then this would be kept in memory and you don't have to recompute it. But pandas users are not familiar with this concept.

A

Next is partition partitioning. Where often, you have to group your logic put data that belongs to a logical group on the same partition, and this often involves shuffling of the data across your cluster.

A

So this is something that pandas doesn't have an interface for and doesn't uh handle well in pandas, even that which is very heavy, which is very reliant on the index. You often find that people uh just use this index as kind of a global um way to to locate data, but that that doesn't hold true in a distributed setting and, of course, testing, because you have to add all of these like extra functions, to bring pandas code to spark or even python code to spark. You normally have to add a lot.

A

This is for one function. You have to add minimum of two helper functions and for each of these functions that you introduce, you have to write additional tests and you have. It becomes a lot harder to to test this, because it's very coupled to these boilerplate functions and that's why fugue was created. So fugue is an abstraction layer for distributed compute.

A

The goal of fugue is to number one, make it easy to use, spark and ask, and also number to unify the inconsistencies that are present between these engines. So in field we, as han showed we want people to be able to describe their logic in python or pandas or sql, and then bring it to fugue and choose the execution engine. So you can define your workflow in python or pandas and then during runtime choose. I want to run this on spark.

A

I want to run this on desk without significant code change and there's interesting properties that uh that come when when we can decouple logic and execution so without fugue, when you use pandas, your pandas code is, is tied to the pandas execution engine and when you use pi spark code, your pi spark code is also tied to the spark execution engine, but when you can decouple logic and execution through fugue, what happens is that you can just define your logic once and then choose where to run it.

A

So often you find that there are a lot of projects where you know you, you started using pandas and the data became too big, and now you need to introduce spark and maybe, instead of going to spark you'll just vertically scale your infrastructure or, on the other hand, maybe you're using spark uh to optimize too early for like a data set as a data set that doesn't need spark right. So now with fugue.

A

You just need to write your code once and define your logic once and then you can choose the execution engine that makes sense during runtime, so you can start small with pandas and then you can move to spark uh so here. In this example, we have a native python python function or pan it uses pandas, and what we can see is that when we define the function, there's no spark dependency whatsoever, and this is what han demoed and once you use the fugue transform function and specify the execution engine.

A

All of this is brought into spark for you.

A

What this did on this slide. I use a pandas data frame but for the same function we can also use native python using the list and dicks, um and we can just uh loop through this list of dicks and apply the pyth native python code uh for each row and again this can also be brought into spark through the transform function so by decoupling logic and execution. You're able to accelerate your testing because you you can test on smaller data with with pandas size data and then choose the execution engine on spark.

A

When you need it you can you also need to test less code in general, so you can accelerate development and then you avoid framework lock-in.

A

So today we're using spark and desk, but if ray starts to pick up and a lot of people start to use ray or maybe even uh the qdf with rapids.ai, where you have gpu clusters, then we can make make an execution engine for those different frameworks and your same python code should be able to map through fugue and because of because you use native python code, that's very testable, there's a lot less maintenance, there's no spark expertise required to maintain the code.

A

Logic as han showed where he had an example of business logic that changed over time. It was very easy to just still use the transform function with fugue, and then we have us. Fugue also has a sql interface, so maybe sql lovers or bi, analysts or data analysts. These these personas can also uh harness the power of distributed compute fugue sql supports spark desk and pandas, also and and then we also support blazing sql uh through um uh to to operate on top of the uh gpu decodif.

A

So with with uh with the sql interface, what han showed in the notebook we have syntax highlighting already implemented, we added keywords to bring sql into the distributed. Setting, for example, we added keywords like load: save persist, partition and transform.

A

So now, sql can be a first class interface where you can um before you would have a sql code, pre sandwiched by predominantly python code, and now you can have it the other way around, where you have predominantly sql code. That invokes python code occasionally, and this is the example of the notebook extension where we have syntax highlight highlighting implemented.

A

So as before I open for questions.

A

I just want to conclude here by saying that number one fugue is also is a mindset first of all, and that our mindset is that we should adapt to the user and allow them to express their logic in whatever way they want to, and then they can express their logic in a scale agnostic way and we can take care of bringing it to a distributed setting when, when you need to scale and we value readability and maintainability of the code rather than deep, framework-specific optimizations, but normally fugue just uses the mechanisms under the hood of that execution engine so that you don't lose a lot of performance.

A

um And with that I I uh I just want to give a quick recap before um so number one. We are an abstraction layer for distributed compute, it adapts to the user, lets them, define their code in native python or sql, and then bring it to spark or desk when needed, and because of decoupling logic and execution. You find that we find that a lot of fugue users can accelerate their big data projects and fugue is just one component of the broader fugue project.

A

So we have few and fugue sql, but we also have fugue tune, which is an abstraction layer for hyper parameter optimization and we also have fugue validate. We also can use fugue to perform data validation by wrapping around pandera or great expectations.

A

I just have contact info here and with that I'll open the floor for questions. Thank you.

A