Hi everyone, my name is Elliot Brossard,
and I'm a Principal Software Engineer at Snowflake.
I'm a tech lead with the Snowpark team,
and today we'll be going behind the scenes
to look at some of the details
of how UDFs work inside of Snowpark.
First we'll cover Java functions, then Python functions,
and finally, we'll take a look
at some best practices for both.
All right, let's get started.
First, we'll take a look at Java functions.
When you use Java functions in Snowflake,
you have a choice of using either UDFs or UDTFs.
Both of these make it possible to transform data
using Java right inside of Snowflake.
When you use a UDF,
you're returning one value as output for every row of input.
With a UDTF, you're returning any number of rows as output
for every row of input.
We also support using Java store procedures too.
This means that you can create and query tables
and other objects using Java,
and you can bring your Java-based workloads into Snowflake.
Java UDFs, UDTFs and store procedures
are all generally available on AWS and Azure
and are in preview on GCP.
What's exciting about Java functions is that they leverage
the existing Snowflake warehouse model for processing.
This means there's no additional cost
to run Java code inside of Snowflake
beyond the compute credits that you pay for normal queries.
The way that we handle Java execution
is that each worker node runs an instance
of the Java virtual machine inside of a sandbox process.
The JVM instances that we'd run
last only as long as query execution does,
and we throw them away at the end of the query.
JARs and other imports that you use in your Java UDFs
use local SSD cache, just like with tables,
so you get a benefit
from reusing the same JARs across queries.
Let's take a look at an example of a Java UDF.
The first thing that you'll do
when you want to create a Java UDF
is that you'll create a JAR from your Java code,
or you can use a CREATE FUNCTION statement
with inline code if it's simple.
Once you've created your JAR,
you can copy the JAR and other imports to a Snowflake stage.
You can use put commands from SnowSQL,
like I've written here,
where I copy files for my local machine
onto a Snowflake stage.
Once I've copied these files, I can create my UDF.
In this case, we're creating a Java UDF
that returns the sentiment such as happy or angry
for a segment of text.
The first thing that we'll do
is we'll specify CREATE FUNCTION,
then the function name itself.
In this case, we've called it DetectSentiment.
The input that we're passing in is text,
which is the string,
and we're returning a string as output.
Note that here we specified language Java
to indicate that we want to use Java for our UDF.
Next, we specify the imports.
These are the files that we copied to our stage,
so this includes this classifier JAR
that holds the Java code, as well as a sentiment model,
which is a gz file that contains the model.
We specify the handler,
which is the method name that we want to call
and is the fully qualified name of the Java method.
It includes the package name,
which is com.my_company in this case.
The class name, which is SentimentAnalyzer
and detect, which is the method name.
Once we've created our Java UDF,
now we can use it in a query.
In this case, I'm selecting text,
and I'm calling my detect sentiment function on the text,
and I'm selecting from a table named SurveyResponses.
This will return the text as well as the sentiment
for each input from the survey responses.
Let's look at how Java functions execute.
The way that we execute Java functions is at query startup,
we copy JARs and other imports
from the stage to the warehouse nodes,
and we give them read-only access in the sandbox to the JVM.
During query execution, each Snowflake worker process
sends batches of rows in parallel to the sandbox process.
During execution, Snowflake monitors memory usage of the JVM
and may run garbage collection periodically
to clean up loose objects.
Snowflake executes Java functions within a sandbox.
This sandbox prevents network access
and sets limits on memory and CPU usage
to ensure that the resources don't compete
with the rest of the execution of the query.
The sandbox does not allow loading native libraries
or spawning new processes.
If you want to read files,
you can do so by using the IMPORTS clause,
or you can use the unstructured data access feature,
which is currently in preview.
Now let's look at Python functions.
Similar to Java functions, Python UDFs and UDTFs
enable you to transform data using Python
right inside of Snowflake.
We also support Python stored procedures,
so if you want to bring your Python-based workloads
into Snowflake, you can do that using a stored procedure.
The one big difference of Python functions
relative to Java functions, however,
is that we have support for Anaconda packages.
Python UDFs, UDTFs and stored procedures
are in public preview,
and they're available for anyone to use.
Snowflake and Anaconda have set up a partnership
and established a Snowflake Conda channel
to enable you to use a curated set of packages
both locally and in Snowflake.
We handle dependencies for you.
We've integrated the Conda package manager
right into Snowflake secure sandbox,
so that you don't need to worry
about managing dependencies yourself.
We've set this up to be scalable and secure.
We've integrated Conda packages inside of Snowflake
using a secure sandbox,
and we provide this at no additional charge
beyond the usual warehouse usage costs.
As with Java functions, Python functions also leverage
the existing Snowflake warehouse model for processing.
There are no additional costs
for using Python functions inside of Snowflake.
It's just the compute credits that you would usually use
for your compute warehouses.
Python functions may use packages
from the Snowflake Anaconda channel, however,
which is updated regularly.
When you create a Python function,
Snowflake solves for the specific packages
that you've specified to determine which to install
in your warehouse prior to query execution.
And we cache these packages just like we do
with table and function imports.
When you create a Python function,
you specify the arguments that you want to use.
We have a specific SQL to Python mapping
for each SQL data type that Snowflake supports.
As an example, if you use a SQL NUMBER type,
and you use a scale of zero, we encode this as an int.
However, if you use a SQL NUMBER type with a non-zero scale
we encode those as a decimal.Decimal value.
You can see the documentation for a complete list
of Snowflake datatype mappings for Python UDFs.
When you want to create a Python UDF,
you can query the INFORMAION_SCHEMA.PACKAGES view
to see a list of the available packages.
As an example, if you want to see
the latest available versions
of NumPy, Pandas, and XGboost,
you can run a query like this.
The output shown here was up to date
as of the time of this writing
but may have changed in the meantime.
What I'm doing in this query
is I'm selecting the package name and the latest version
for each of the packages in this list.
Note that I've specified the language is Python
and the package names that I'm looking at
are in NumPy, Pandas and XGboost.
When you want to use Anaconda packages in a Python function,
you can specify a list of packages in this packages clause.
Here I specified that I want NumPy, Pandas and XGboost.
Let's take a look at how I've used them.
In the packages list, I've specified that I want Numpys.
I've used equals, equals here
to specify a specific package version for Xgboost,
but note that for Pandas, I've specified a wild card
indicating that I'm okay with matching any version of Panda
as long as it starts with 1.4.
If there's no combination of package versions
that match the criteria I specified, however,
the CREATE FUNCTION statement will fail.
When I run a query that uses a Python function,
these packages are all installed seamlessly.
If I want to look at the solved packages
for a particular UDF
I can use the Describe statement.
Here, if I describe py_udf,
what I'll see is that the packages I specified
were NumPy, Pandas and XGboost,
but the installed packages looks like this huge list.
In this output,
I've highlighted a few of the interesting aspects here.
You can see that there's this huge list
of dependencies that were pulled in,
but the specific versions that I got for NumPy was 1.2, 1.5.
For Pandas, it was 1.4.1, and for XGboost, it was 1.5.0.
Something to keep in mind when you create Python UDFs
is that the solved packages are frozen.
What this means is that when you create a Python function,
the specific versions of packages
are determined at that point and will not change,
even if new packages are released.
You can use the GET_DDL statement
to return the original DDL that you used.
In this case, if I use GET_DDL,
and I specify function and py_udf,
the output will give me back the original specification
of the packages that I used.
I can run the CREATE OR REPLACE FUNCTION statement
that is returned, and it will update the solved packages,
such as to pick up a new release of Numpy.
Now let's take a look at Python function execution.
When we start a query that uses a Python UDF,
the packages that were specified
are installed on the warehouse nodes.
If I recently ran another query
that used the same packages on the warehouse,
they will be cached and do not need to be installed again.
The files that I've specified in my imports list,
which could include .py files, .zip files or other imports
are either downloaded or pulled from the cache,
so I also get a benefit from reuse
if I run the same query multiple times.
Since Python has a global interpreter lock,
we create many Python interpreter processes
for each function that's used in the query.
We try to be as efficient as possible
by initializing the Python interpreter
before forking additional processes
in order to reduce initialization time.
When the query is done,
we keep packages and other imports around in a local cache,
but the sandbox and Python interpreters are cleaned up.
If you rerun the same query or in another query
using some of the same packages or imports,
because these are cached, the queries will be faster.
As with Java functions, we also sandbox Python functions
to prevent network and local file system access.
To keep up to date on common vulnerabilities
and exposures for Python packages,
you can look at Anaconda's website.
The good thing though,
is that even though there may be CVEs for some packages,
the sandbox helps to mitigate them.
Because we prevent network access
and local file system access,
we can prevent many causes of attacks
that could lead to data exfiltration
or attacks on the host system itself.
Now let's look at some best practices
for Java and Python functions.
First, a note on parallelization.
When you run a query that uses a Python or Java UDF,
Snowflake attempts to use the full power of your warehouse.
What this means is that we redistribute rows
between nodes in the warehouse
to help parallelize expensive computations.
We also collect some historical statistics.
We try to see if your UDF is fast or slow
and use that to determine
how we distribute the work among workers
in order to make your query as fast as possible.
If you use a LIMIT clause in your query, however,
or a heavily skewed GROUP BY, PARTITION BY or JOIN,
that may prevent our ability
to effectively parallelize your query.
You may want to use the Python UDF batch API.
If you are using libraries such as XGboost or PyTorch
that operate efficiently on batches of rows.
There is no need to change your SQL
to take advantage of the optimized Python execution.
You can do it entirely in your Python code.
When you use the batch API,
the result of your UDF should be a Pandas Series or array,
a NumPy ndarray, or a Python sequence.
We optimize encoding and decoding more possible,
such as for arrays of floating point and integer values.
Here's an example of a Python UDF
to define using the batch API.
You'll see that I've defined this scoring function,
and I'm taking this model file and model loader as input.
The packages that I'm using are NumPy, Pandas and XGboost.
The handler that I'm going to use is score,
which is defined below.
So the interesting parts here
are that I've specified this vectorized attribute,
and you can see that I've specified
that I want to use Panda's data frame
as the input representation.
The target batch size is a thousand.
This means that I'm hoping to get a thousand rows of input
in each data frame that I receive.
Now let's take a look at the scoring function.
It's going to One-hot-encode categorical attributes,
and then it's going to join the results of One-hot-encoding
to the rest of the attributes specified as input.
Once it's done that,
it can run this prediction function
using this entire batch of a thousand rows.
This is much faster over a batch of rows
as opposed to a single row at a time.
We've seen huge improvements when using the batch API
of up to a hundred times performance.
Let's take a look at how the batch API works
behind the scenes.
Similar to scale UDF execution, we send batches of rows
from the worker process to the Python UDF server.
In this case, however, we convert to a Panda's DataFrame
and then call your handler method with it.
We take the result, and we convert from a series
back into a batch of rows,
and then send that back to the worker process
for further processing.
Now, some best practices on Java on Python UDTFs.
One thing to keep in mind when defining your UDTFs
is that you should try to emit output lazily.
For Java UDTFs, this means using a stream,
and for Python UDTFs it means using a sequence.
If you return a stream or a sequence, it enables Snowflake
to continue processing as rows are returned,
and it avoids needing us
to hold all results in memory at the same time.
Let's take a look at a Java UDTF example.
So this is a bad example.
This is something you should not do in your own code.
The idea of this code is that I'm extracting tags from XML
and I'm building up this entire array,
and then finally I'm using Arrays.stream
to return those results.
What's bad about this is that I need to extract
all of the tags before I return any of them.
Ideally, what I should do instead
is I should return a stream of them
as I find each tag in the XML.
So here's a good example.
What I'm doing instead here
is that I'm returning matches as they're found in this XML.
I've built up this stream, which is evaluated lazily,
calling this helper method
that returns the tags one at a time as I find them.
In this Python UDTF example,
I also have an XmlTagExtractor.
I've defined this method that finds the next tag,
and what I'm doing here is I have a while loop
that's going to find all the tags to build a list
and then return them all at once.
This is inefficient though.
I have to find all of them before I can return.
What would be better, similar to my Java UDTF example,
is that I return them as I find them.
So let's take a look at a good example instead.
In this good example what I'm doing
is that as I find each match,
I'm calling yield to return results as they're available.
Something to keep in mind
when working with Java and Python UDTFs
is that Snowflake constructs an instance of the handler
for each partition that it sees.
If you want per-partition state, such as a running sum,
you can initialize that in the constructor.
If you want to have common state, however,
such as an XML parser or ML model,
you should use static variables
for Java UDTFs or a static block
and do initialization once per ClassLoader.
For Python UDTFs, you should use module-level initialization
to initialize once per Python interpreter process.
Let's take a look at how to put this in practice
using a Java UDTF.
You can see here that I've defined
this private static final variable called XmlParser,
and this is going to be initialized
only once per ClassLoader.
If I had instead made this just a private final variable
that was initialized in the constructor,
it would mean that I was creating an XmlParser
for every partition that Snowflake sees.
For Python UDTFs, we can do something similar
where we put the XmlParser at module level scope.
This means that it will be only initialized once,
even though our XmlTagExtractor class
will be constructed for every partition that we see.
Now a note on using temporary storage
from Java and Python UDFs.
As I mentioned previously,
the sandbox prevents local file system access.
What we do allow, however, is access to temporary storage.
What this means is that if you want to have some files
that you create for working contents of your UDFs,
you can write them to the temp directory.
For Java UDFs, if you want a unique temporary directory,
you can use Files.createTempDirectory.
In Python, you can also create a temp directory
or you can create a UUID
to use this part of the temp directory path.
If you want to have a shared directory
among local instances of the same function,
what you can do in Java is to use a static block
or variable assignment to create the directory.
In Python, it's a little bit trickier.
You can use a pattern like this, however,
where you have a lock file to synchronize access.
Here I've defined a method called acquireLock,
and what this will do is it's going to open a file
within the temp file system, called myfunctionlock.LOCK.
Once I've acquired this lock,
I can perform whatever initialization I want to
on the file system and then release it
so other instances of the same UDF can proceed.
Now a bit on logging.
When you write Java and Python functions,
you can log to event tables.
This is a very powerful functionality
that can help with debugging
and also tracking errors and custom statistics.
After you've run a query using a Java or Python function
that performs logging,
you can query the event table to see the logs.
For Java user logging, you can use the slf4j.Logger,
and you can instantiate it
as a static variable in your class.
Then you can call logger.info or logger.debug
or whatever level that you want
to log information during execution.
From Python UDFs, similar idea.
I can import the logging module and then I can log
whenever I think something interesting has happened.
In this case, what I want to log
is that the divisor was zero for some input.
I can query my event table after I've run a query
that hits this exception
and try to figure out what went wrong.
Thanks so much for watching.
Please be sure to ask your own questions
and share best practices on the Snowpark Forums.
If you enjoyed this content, give it a like
and hit the subscribe button for additional content.