- Hi, welcome everyone.
Today we're going to be talking about optimizing Spark UDFs.
I hope everyone's staying safe
and healthy.
Before we start the session let me introduce myself.
My name is Shivangi.
I'm a Senior Engineering Leader at Informatica.
I have about 10 + years of experience dealing
with database systems, Distributed computing
and other big data technologies like Spark, Hadoop etc.
So let me just give a brief background
about what Informatica is.
So Informatica is one
of the leading providers of data engineering solutions.
We have a suite of offerings ranging
from data integration solutions to cloud data management,
data quality, data governance, etc.
For all of these offerings, we do use Spark extensively.
We were one of the early adopters of Spark.
So we started with Spark 1.6,
and we are at Spark 3.0 right now.
As we started solving our use cases
we took the open source version
of Spark and really innovated
on top of it to solve our different kind
of data engineering specific use cases.
All right, so here's the agenda of the session today.
We are going to start with the introduction to UDFs.
We'll walk through certain examples, benefits of UDFS.
Then we will talk about performance considerations.
And that's where the optimization part comes in.
We'll talk about the suggestive
alternatives that we can take to optimize UDFs.
And then finally look at some performance results.
So what are UDFs?
UDFs are User-Defined Functions
which basically all give you the
ability to extend Sparks language.
It is something that you can write a custom function
for registered against the Spark language, and then use it
as a native function across your Spark programs.
So a typical example of UDF, how you would define that,
here I am for simplicity, defining something
really, really simple.
I just want to define the plusOne operation.
So I define a UDF by just saying
this is a scholar function that does X+1 operation.
I go ahead and register it against a
UDF registration framework.
I give it an alias.
So here I provided the alias as plusOne.
And that is how you would recognize the function.
And then you can go ahead and use it in your language.
You can say select plusOne and you'll get the results
for that UDF.
So what are the benefits?
As you saw it helped us extend the built-in capabilities
of Spark SQL.
It's very simple and straightforward to implement.
So I didn't have to do much to implement the operation.
All of the complexity just lies
within the function that you're trying to implement.
It does come with a plug and play architecture.
So you build it once
and then you can use it across different Spark programs
and also the API's
are pretty stable, so they are backward compatible.
So if you're moving
from one Spark version to another Spark version
we don't have to worry about changing these UDF's
or your Spark program specifically for these UDF's.
Now at Informatica we extensively utilize UDF's.
We have our own expression language
and these functions are not available natively with Spark.
So we decided to go the UDF route.
But as we started looking
at the performance of our application
we realized that there were several performance bottlenecks
that with there with UDF's.
Spark optimizer has evolved over the period
of time from like 1.6 version to 3.0.
There are a lot of optimizations which have been added
but we could not utilize those optimizations
because of fusing UDF's.
So anything like WholeStageCodegen, Null optimization,
Predicate Pushdown, everything was getting blocked.
Another thing that we observed
was Spark tends to treat its strings as UTF-8.
But when you exercise a UDF, it gets converted
into Java runtime, which is using your java.line.string,
which means that it has UTF-16 encoding.
And we were going back and forth between UTF-8
and UTF-16, which was an expensive operation.
So any time you had seen input
or a string, a written type, it added cost to your UDF's.
So let's look at some of these things in play.
So here I'm trying to analyze the physical
plan and see how Spark is actually
implementing a native operation versus a UDF.
So if you see I'm doing a very simple operation
PlusOne operator.
I select the value and it's pretty unremarkable.
I do the same thing with PlusOne.
And the physical plan looks almost exactly the same,
but that's because the PlusOne operation here,
Spark did not have room
for a lot of optimizations.
So we want to get a little more creative
and see when the Sparks optimizer comes into play.
So here I'm trying to do the same PlusOne operation
but I'm applying a filter on top of my dataframe.
So I do dataframe.filteroperation.
I use my X+1 operator and say,
X+1 should be equal to two, very simple.
But when you look at the plan, you see
the highlighted portion and that highlights
that there is a pushed filter in my physical plan.
So Spark was actually able to push down a predicate.
Now let's analyze the same thing with PlusOne.
So I'm exactly doing the same operation.
I replaced my X+1 with PlusOne UDF that we just created.
And if you look at the physical plan
and we are specifically looking at the highlighted portion
we see that Spark was unable to push down any filters.
So this is a simple operation, but as your queries go
om to be more complex,
this performance cost adds up
and the UDF's starts slowing down Spark programs.
So looking at this, we decided that we need
to redesign our UDF's to make our programs more performant,
but we wanted to keep the benefits of UDF's,
but still get the performance benefits
that we get from Spark optimizer.
That was our main design goal.
We wanted to get these capabilities with minimal changes
but still treat UDF's as native Spark functions.
So let's walkthrough what we decided to do at Informatica to
make these UDF's more efficient.
So here, if you're not aware, the expression framework
from Spark comes from the SQL catalyst project.
So what we wanted to do since we wanted
to maintain our own isolation, we created a new project.
So you create a completely new project, which we call
Infaexpressions here.
The only thing you want to do is maintain
a project structure, which is similar to your Sparks
SQL catalyst.
So if you see, we follow the same packet structure,
you define that
and I define all my expression classes
inside of that packet structure.
The next thing to do is for us to be able to make Spark
understand that anything that
I'm implementing is actually an expression
in Sparks language.
So how does Spark understand that?
For Spark to be able to recognize anything as an expression
we need to extend it from their base expression class.
So there is a base expression class
under your org.apache.spark.sql.catalyst project,
and it has certain implementations
which are already available
for you to use like UnaryExpressions, TernaryExpressions,
these are for like single argument,
multi argument expressions.
So you can even extend from that.
What we are looking for basically
is to satisfy the expression contract that's defined
by Sparks expression class.
So here you might want to look
at these two methods which are there,
which are doGenCode
and eval method.
These are the two
things that we have to specifically
define, to define the functionality that's represented
by this UDF.
If you're curious, Spark has all
of its functions under this following location,
you can go to the Spark code under the SQL catalyst project.
Look here, and there are a lot
of samples available to like get inspiration
on how these functions can actually be implemented
which are like really performant.
So keeping that in mind, we decided to re-implement
our function.
So here, for example, I'm going to re-implement
the same thing.
The PlusOne that we were doing before,
I want to make it a native function.
So I'm going to use this as PlusOneNative.
I'll call that as PlusOneMative for simplicity
and distinction, I create a class which is PlusOneNative.
It takes them a single argument
so I extend it from the UnaryExpression class.
And also I may get null in tolerant,
so that's something that Predicate Pushdown looks for.
So these are the kinds of constraints that you have to look
for if you're going after a certain particular optimization.
This is similar to how Spark would identify
even for native functions.
So we are not adding any expansion
to the functionality, but trying to understand Sparks logic
of applying these optimizations.
I go ahead and I re-implement
like the doCodeGen method and nullSafeEval.
So with UnaryExpression, you can implement Eval
or nullSafeEval.
I chose to implement nullSafeEval.
So I had that here.
So once I have done that,
the next step in the process for me after
defining this class is for Spark to recognize
that this is indeed a function.
So I need to create a registry for it.
I created a function definitions file.
All of these functions definitions file
for Spark today exist under the same project
and it's called the functions.scala.
So I created a similar file
which is Infafunctions, and I defined all
of the functions that I'm planning to implement.
So if you see, I have PlusOne native here.
So this is the name that would be used
by Spark to recognize the function.
And I actually, along with this,
give it the definition of what it's implemented by.
So if you see, my PlusOneNative class
is what defines this function.
Once I've done that, I'm actually done
implementing my UDF.
All I now need to do is combine
and make Spark aware of this jar.
So you compile your project,
and the next thing to do is to make Spark aware
that this jar exists.
The simplest way to do that is to add this
in the Spark assembly.
You can go to the Spark/jars folder
and drop in your jar there.
Once we drop in our jar there, we are good to go.
So now having done that
I can just go to my Spark shell or write a Spark code,
all I need to do is import this function.
So if you see, I imported infafunctions.plusOneNative,
and I was able to utilize this function
as any other native Spark function.
A lot of us do use SQL queries for Spark SQL.
And if I'm doing that, I need Sparks SQL parser
to be able to understand that this expression exists.
And that knowledge actually comes
from the FunctionsRegistry.scala class.
This function registry again exists
under the SQL catalyst project.
If you have to use SQL queries
you will have to edit this file.
So this is only a small change that you have to make
into the Spark code.
So how would we do that?
We download the Spark code for the
version that you're using,
and we go under the Spark catalyst project.
We find the file Functions.Registryscala
and register our function.
So if you see, I have added my function.
You'll say that this is the function,
it's defined by PlusOne native class,
and I give it an alias which is again, PlusOneNative.
That's how the SQL query is going
to recognize that function.
Now I again need to recompile my Spark code.
The one thing to note here is when you're recompiling
you need your catalyst project to be aware
of this new project that you've added
which actually has the definition for PlusOneNative.
So I edit the pom file, add a new dependency,
build this pack class Spark catalyst jar.
And once we have the new jar, we do the exact same thing
of going under the Spark assembly,
and now I replace the existing Spark catalyst jar
with my new jar.
Once I have done that, that's all the steps
that I need to do.
So once we add it to the functions registry
I can go back to my Spark shell, execute a query that uses
PlusOneNative function.
So if you see, I'm trying to do a select
PlusOneNative from a table and I get the results.
So Spark was able to actually recognize PlusOneNative
as an existing function and give me the results.
So now that we've done all of this,
we go back to our original point.
The reason we wanted to do all of this
was to utilize Sparks optimization.
So now let's see
is Spark able to actually optimize these native functions.
So I do the same operation that I did and where I showed you
that UDF was not able to push any predicates
down.
I filter my data frame.
I apply the same operation
which is PlusOneNative=2,
and I look at my physical plan.
And here we see the highlighted portion shows
that I was actually able to push
down the same filter that the native
PlusOne operator was able to do.
So this kind of proves that yes,
Spark is able to use these functions now
and use them as native functions
and allow them to participate in their optimization step.
So let's talk about some more tips and tricks.
Also, along the way we realized that, you know
when we are writing these functions
we want to be especially careful
about how we are coding these functions
because that does contribute to the performance overhead.
We should definitely consciously make
an effort to avoid temporary object creation.
It's not apparent with smaller data sets,
but as your data set size grows,
you start seeing the effect of garbage collection
on temporary objects.
So it starts becoming
as an overhead to your performance.
On the same lines,
we should try using Scala's while construct over for,
because for will cause temporary object allocation.
We should consider using imperative style
over functional style.
The other thing is you might consider
using thread static variables to allocate temporary buffers
and that's to avoid synchronization issues.
So I didn't even talk about the UTF-8 to UTF-16
conversion initially
which was like a significant overhead for string operations,
but sometimes this conversion is actually necessary.
You cannot avoid it, but in those cases
you might want to actually consider
lazy conversion instead of actively
converting all strings from UTF-8 to UTF-16.
You might want to inspect your strings
and then only convert.
That also, I guess significant performance benefit.
So having talked about all of the
things that we can do to improve the UDF performance
let's look at some of the performance results.
So here I'm looking at a string function.
This was our re-implementation of the substring function.
Informatica expression language defined substring
in a slightly different way from what Spark has.
So our contracts were different.
So we decided to re-implement it and we re-implemented it
keeping all previous things in mind,
here the orange bar is a UDF
and the lined bar is actually when we converted it
into a native function in Sparks language.
As you observe, and I'm going from scale factor one to 50,
so we are looking at one gigabytes to 50 gigabytes of data.
When the data set was smaller
the performance was still better,
but it wasn't as apparent, but as your data size
starts growing and 50 gigabytes is not a lot of data.
You see that the performance is considerably faster
like an operation as for our program that was taking
18 minutes, now takes just six minutes for us to execute.
Let's look at some more performance numbers.
So here I'm looking at the date function,
again here we were trying to do add to date
to date kind of operation.
Hear the rewriting specifically
considered avoiding temporary object creation.
So we used a lot of wire loops
and avoided any temporary object creation.
And we observed that, like for larger data sets
we could actually see a 100 percent performance gain
for our data sets.
And finally here's some numeric function.
we just re-implemented the rate function
for us because that comes from our expression language.
And this was already
like an efficient function, but just making it native
like gave us about 50% of performance boost.
So as you can see, like if we start utilizing a lot
of these things that we have talked
about, we can get significant performance gains.
Like the performance results that we were looking
at were for very simple operations.
We were just trying to evaluate the performance
of that particular UDF
getting converted into a Spark native function.
But if you have a lot of these UDF's
in your Spark program and your program becomes arbitrary
large, you will see even faster performance
than what we've seen in our performance benchmarking.
All right, so with that I conclude my talk
and I'll leave the floor open for any questions,
feedback that you may have.