Pandas UDF and Python Type Hint in Apache Spark 3.0

In the past several years, the pandas UDFs are perhaps the most important changes to Apache Spark for Python data science. However, these functionalities have evolved organically, leading to some inconsistencies and confusions among users. In Apache Spark 3.0, the pandas UDFs were redesigned by leveraging type hints. By using Python type hints, you can naturally express pandas UDFs without requiring such as the evaluation type. Also, pandas UDFs are now more ‘Pythonic’ and let themselves define what the UDF is supposed to input and output with the clear definition. Moreover, it allows many benefits such as easier static analysis. In this talk, I will introduce the redesigned pandas UDFs with type hints in Apache Spark 3.0 with a technical overview.

About:
Databricks provides a unified data analytics platform, powered by Apache Spark™, that accelerates innovation by unifying data science, engineering and business.

Connect with us:

[ZoltanCToth-sbqo]

This has been among the most concise technical explanation videos I've bumped into in the past years. Thanks for sharing!

[haneulkim]

Amazing talk. At @18:30 about iterator of series to iterator of series pandas_udf the code initialized (say load model) here. My code is in pyspark.2.4.7 and trying to migrate to 3.x. Originally model was loaded within the script above the pandas_udf code, even though model is loaded it still gets broadcasted each time I call pandas_udf? So my question is which is better between initializing model in above then than simply referencing within pandas_udf or initializing the model within pandas_udf function.

[chinmoyjyotikathar]

The errors thrown by pandas udf functions are really hard to debug and make sense of. Suppose I define the schema for the return dataframe for a group map pandas function, the error thrown for type mismatch for some particular column does not even state the column name. You'd have to go through all the columns one by one and change the type of each and rerun in worst case scenario.

[gouravchoubey]

Hi,
I trying to execute below code snippet for pandas_udf to calculate quantile.The udf returns as numpy float.

My spark version is 2.4.3

from pyspark.sql.functions import col, lit, pandas_udf, array, when, PandasUDFType
import pandas as pd
import numpy as np
from pyspark.sql.types import FloatType
columns=['A', 'B', 'C', 'D', 'E', 'F']
df=sc.parallelize([(60.0, 60.0, 45.0, 45.0, 45.0, 45.0), (12.0, 17.0, 18.0, 20.0, 23.0, 28.0), (11.0, 20.0, 24.0, 23.0, 45.0, 55.0), (0.0, 0.0, 0.0, 0.0, 0.0, 0.0)]).toDF(("A", "B", "C", "D", "E", "F"))

@pandas_udf("float")
def quan(row)->float:
print(row)
print(type(row))
value = row.iloc[0]
print("Value : ", value)
print("Value is ", type(value))
quan=np.quantile(value, 0.25)
return quan

df=df.withColumn("q25", quan(array(columns)))
df.show()

But I get below error?

TypeError: Return type of the user-defined function should be Pandas.Series, but is <class 'float'>

Although I have mentioned return type as float