Apache Spark™ 3.5 and Databricks Runtime 14.0 have introduced an thrilling characteristic to the desk: Python user-defined desk capabilities (UDTFs). On this weblog publish, we’ll dive into what UDTFs are, why they’re highly effective, and the way you should utilize them.

What are Python user-defined desk capabilities (UDTFs)

A Python user-defined desk perform (UDTF) is a brand new form of perform that returns a desk as output as a substitute of a single scalar outcome worth. As soon as registered, they will seem within the FROM clause of a SQL question.

Every Python UDTF accepts zero or extra arguments, the place every argument could be a fixed scalar worth resembling an integer or string. The physique of the perform can examine the values of those arguments with a view to make choices about what knowledge to return.

Why do you have to use Python UDTFs

Briefly, if you would like a perform that generates a number of rows and columns, and need to leverage the wealthy Python ecosystem, Python UDTFs are for you.

Python UDTFs vs Python UDFs

Whereas Python UDFs in Spark are designed to every settle for zero or extra scalar values as enter, and return a single worth as output, UDTFs supply extra flexibility. They’ll return a number of rows and columns, extending the capabilities of UDFs.

Python UDTFs vs SQL UDTFs

SQL UDTFs are environment friendly and versatile, however Python provides a richer set of libraries and instruments. For transformations or computations needing superior strategies (like statistical capabilities or machine studying inferences), Python stands out.

create a Python UDTF

Let’s take a look at a primary Python UDTF:

from pyspark.sql.capabilities import udtf

@udtf(returnType="num: int, squared: int")
class SquareNumbers:
    def eval(self, begin: int, finish: int):
        for num in vary(begin, finish + 1):
            yield (num, num * num)

Within the above code, we have created a easy UDTF that takes two integers as inputs and produces two columns as output: the unique quantity and its sq..

Step one to implement a UDTF is to outline a category, on this case

class SquareNumbers:

Subsequent, it’s essential to implement the eval technique of the UDTF. That is the strategy that does the computations and returns rows, the place you outline the enter arguments of the perform.

def eval(self, begin: int, finish: int):
    for num in vary(begin, finish + 1):
        yield (num, num * num)

Word the usage of the yield assertion; A Python UDTF requires the return sort to be both a tuple or a Row object in order that the outcomes may be processed correctly.

Lastly, to mark the category as a UDTF, you should utilize the @udtf decorator and outline the return sort of the UDTF. Word the return sort should be a StructType with block-formatting or DDL string representing a StructType with block-formatting in Spark.

@udtf(returnType="num: int, squared: int")

use a Python UDTF

In Python

You may invoke a UDTF immediately utilizing the category identify.

from pyspark.sql.capabilities import lit

SquareNumbers(lit(1), lit(3)).present()

+---+-------+
|num|squared|
+---+-------+
|  1|      1|
|  2|      4|
|  3|      9|
+---+-------+

In SQL

First, register the Python UDTF:

spark.udtf.register("square_numbers", SquareNumbers)

Then you should utilize it in SQL as a table-valued perform within the FROM clause of a question:

spark.sql("SELECT * FROM square_numbers(1, 3)").present()

+---+-------+
|num|squared|
+---+-------+
|  1|      1|
|  2|      4|
|  3|      9|
+---+-------+

Arrow-optimized Python UDTFs

Apache Arrow is an in-memory columnar knowledge format that enables for environment friendly knowledge transfers between Java and Python processes. It will possibly considerably enhance efficiency when the UDTF outputs many rows. Arrow-optimization may be enabled utilizing useArrow=True.

from pyspark.sql.capabilities import lit, udtf

@udtf(returnType="num: int, squared: int", useArrow=True)
class SquareNumbers:
    ...

Actual-World Use Case with LangChain

The instance above would possibly really feel primary. Let’s dive deeper with a enjoyable instance, integrating Python UDTFs with LangChain.

from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from pyspark.sql.capabilities import lit, udtf

@udtf(returnType="key phrase: string")
class KeywordsGenerator:
    """
    Generate a listing of comma separated key phrases a few subject utilizing an LLM.
    Output solely the key phrases.
    """
    def __init__(self):
        llm = OpenAI(model_name="gpt-4", openai_api_key=<your-key>)
        immediate = PromptTemplate(
            input_variables=["topic"],
            template="generate a few comma separated key phrases about {subject}. Output solely the key phrases."
        )
        self.chain = LLMChain(llm=llm, immediate=immediate)

    def eval(self, subject: str):
        response = self.chain.run(subject)
        key phrases = [keyword.strip() for keyword in response.split(",")]
        for key phrase in key phrases:
            yield (key phrase, )

Now, you may invoke the UDTF:

KeywordsGenerator(lit("apache spark")).present(truncate=False)

+-------------------+
|key phrase            |
+-------------------+
|Massive Knowledge           |
|Knowledge Processing    |
|In-reminiscence Computing|
|Actual-Time Evaluation |
|Machine Studying   |
|Graph Processing   |
|Scalability        |
|Fault Tolerance    |
|RDD                |
|Datasets           |
|DataFrames         |
|Spark Streaming    |
|Spark SQL          |
|MLlib              |
+-------------------+

Get Began with Python UDTFs As we speak

Whether or not you are trying to carry out complicated knowledge transformations, enrich your datasets, or just discover new methods to research your knowledge, Python UDTFs are a priceless addition to your toolkit. Strive this pocket book and see the documentation for extra data.

Future Work

This performance is just the start of the Python UDTF platform. Many extra options are at present in growth in Apache Spark to develop into out there in future releases. For instance, it should develop into potential to assist:

• A polymorphic evaluation whereby UDTF calls might dynamically compute their output schemas in response to the particular arguments offered for every name (together with the sorts of offered enter arguments and the values of any literal scalar arguments).
• Passing whole enter relations to UDTF calls within the SQL FROM clause utilizing the TABLE key phrase. This can work with direct catalog desk references in addition to arbitrary desk subqueries. Will probably be potential to specify customized partitioning of the enter desk in every question to outline which subsets of rows of the enter desk can be consumed by the identical occasion of the UDTF class within the eval technique.
• Performing arbitrary initialization for any UDTF name simply as soon as at question scheduling time and propagating that state to all future class situations for future consumption. Which means that the UDTF output desk schema returned by the preliminary static “analyze” technique can be consumable by all future __init__ calls for a similar question.
• Many extra fascinating options!