Skip to main content
Open Source

Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs

Define more performant UDFs with ease.

by Ruifeng Zheng and Yicong Huang

  • We introduce native Arrow UDFs, which operate directly on Arrow data, eliminating the Pandas/Arrow conversion overhead in Pandas UDFs for faster execution and lower memory usage.
  • We also describe Arrow UDF types for scalar and aggregation use cases, and Arrow UDTFs for table-in, table-out transformations, with code examples in both Python and SQL.
  • Benchmarks show Arrow UDFs are ~10% faster and use ~40% less memory than Pandas UDFs, with better support for complex datatypes.

Introduction

Python User-Defined Functions (UDFs) are an essential extensibility mechanism but have traditionally suffered from high overhead due to row-based execution. In Apache Spark™, Pandas UDFs addressed part of this problem by introducing Arrow-based serialization and batch processing, significantly improving throughput compared to scalar Python UDFs.

However, Pandas UDFs still have fundamental limitations:

  • The Pandas/Arrow data conversion introduces additional data copies. Zero-copy approaches are only possible in certain narrow cases. For example, columns with NULL values will trigger deep copies.
  • Complex datatypes are not supported well. For example, nested StructType instances are not supported for the output type with aggregation use cases.
Data flows of Pandas UDF Execution in Apache Spark™

By dropping the Pandas/Arrow data conversion, the Arrow UDFs execute faster than Pandas UDFs, consume less memory, and provide better datatype support.

Native Arrow UDFs

We’re thrilled to introduce Native Arrow UDFs starting with Databricks Runtime 18.0 (release notes), an exciting leap forward for performant UDF execution.

Native Arrow UDFs operate directly on Arrow data without converting inputs into Pandas or NumPy objects. This preserves the columnar layout end-to-end, avoids unnecessary data copies, and lets UDFs use vectorized processing by leveraging Arrow’s native compute and memory model.

To define an Arrow UDF, users are able to use a new python decorator @arrow_udf, with specified return type and optional evaluation type. For instance:

Users can also define it with existing decorator @udf with complete type hints. For instance:

Note: The function definition should include type hints for all of the arguments and the return value.
This design aligns with the interfaces of scalar Python UDFs, providing a consistent and intuitive experience for users already familiar with scalar Python UDFs.

The following demonstrates how to use the Arrow UDF:

Python Usage:

SQL Usage:

We provide support for variants of Arrow UDF interfaces. Including Scalar Functions, Aggregate Functions and Table Functions. In the data frame API we also provide mapInArrow and applyInArrow to use Arrow UDFs. We will next introduce them one by one. 

Arrow Scalar Functions

Arrow Scalar Functions perform row-wise transformations. They are the Arrow equivalent of scalar Pandas UDFs and can be used anywhere a column expression is expected, such as df.select() or df.withColumn(). Three input modes are supported: direct, iterator, and iterator of multiple arrays. The iterator variants are useful when the UDF requires expensive one-time initialization (e.g.,  loading a model or compiling a regex pattern), as the setup cost is amortized across all batches. In all cases, the output row count must match the input row count.

  • Arrays to Array: receiving one or more pyarrow.Array and returning one pyarrow.Array. The input and output array must have the same number of values.
  • Iterator of Arrays to Iterator of Arrays: receiving an iterator of pyarrow.Array and returning an iterator of pyarrow.Array. This type is useful when the UDF execution requires expensive initialization. 
  • Iterator of Multiple Arrays to Iterator of Arrays: receiving an iterator of a tuple of multiple pyarrow.Array and returning an iterator of pyarrow.Array.

Arrow Aggregate Functions

Arrow Aggregate Functions take one or more pyarrow.Array inputs and return a scalar value, reducing a group of rows into a single result. They are the Arrow equivalent of grouped aggregate Pandas UDFs and are used with groupBy().agg() or Window operations. Similar to scalar functions, aggregate functions also support three input modes. 

Arrays to Scalar: receiving pyarrow.Array and returning a scalar value. 

  • Iterator of Arrays to Scalar: receiving an iterator of pyarrow.Array and returning a scalar value. This is useful for processing large volumes of data in aggregation-style operations.

Iterator of Multiple Arrays to Scalar: receiving an iterator of a tuple of multiple pyarrow.Array and returning a scalar value. More complex aggregations can be defined.

Arrow Table Functions

Arrow Table Functions, also known as Arrow UDTFs (User-Defined Table Functions), accept a pyarrow.RecordBatch or multiple pa.Array as input and produce a pyarrow.Table as output. This represents the predominant pattern for table-in, table-out transformations implemented in Python utilizing columnar execution. Arrow UDTFs possess the capability to:

  • Return multiple columns
  • Produce zero, one, or multiple rows
  • Execute vectorized table transformations employing Arrow compute kernels

Consequently, they are optimally suited for operations such as filtering, row expansion, data restructuring, and the generation of derived columns.

The arrow_udtf interface is designed for simplicity, employing a decorator syntax where you define the return type using a DDL-formatted string. In this setup, the eval method takes PyArrow objects as input and is expected to yield PyArrow Tables or RecordBatches. The interface accommodates two input modes. When processing table arguments, the eval method is provided with a pa.RecordBatch object that encapsulates all columns from the input table:

For scalar arguments, the method receives pa.Array objects, one for each scalar input:

Here is another example:

This UDTF can work in two distinct ways:

Python Usage:

SQL Usage:

DataFrame mapInArrow and applyInArrow Support

In addition to User-Defined Functions (UDFs) and User-Defined Table Functions (UDTFs), PySpark furnishes Arrow Function APIs that facilitate the direct application of Python native functions to Arrow data at the DataFrame level. These APIs operate analogously to their Pandas counterparts (mapInPandas, applyInPandas) but utilize pyarrow.RecordBatch and pyarrow.Table instead of Pandas DataFrames, thereby circumventing the conversion overhead between Pandas and Arrow formats.

  • Map. DataFrame.mapInArrow transforms an iterator of pyarrow.RecordBatch into another iterator of pyarrow.RecordBatch, enabling row-level operations such as filtering, transformation, or expansion.
  • Grouped Map. groupBy().applyInArrow() applies a specified function to each group, accepting and returning a pyarrow.Table. This functionality proves beneficial for per-group transformations, such as data normalization.
  • Co-grouped Map. cogroup().applyInArrow() permits the cogrouping of two DataFrames based on a shared key, subsequently applying a function to each cogroup. The function receives two pyarrow.Table inputs and is expected to return a single pyarrow.Table.

Performance

By removing the expensive Pandas/Arrow data conversion, Arrow UDFs generally execute faster than Pandas UDFs, with less memory usage. Let’s compare the two simple UDFs:

The Arrow UDF is ~10% faster than the Pandas UDF, and the memory profiler shows that ~40% memory is saved in the execution.

Conclusion

Databricks Runtime 18.0 introduces Native Arrow UDFs, offering a faster, leaner alternative to Pandas UDFs for performant Python UDF execution in PySpark. By operating directly on Arrow data and eliminating the Pandas/Arrow conversion overhead, Arrow UDFs deliver ~10% faster execution, ~40% less memory usage, and better support for complex datatypes -- all with a familiar, intuitive decorator syntax.

Ready to explore more? Try out Native Arrow UDFs today on Databricks as part of Databricks Runtime 18.0. To get started, simply replace your existing Pandas UDFs with Arrow UDFs. In most cases, it only takes a few lines of change to unlock immediate performance gains. See the Arrow UDF documentation and Arrow UDTF documentation for the full API reference and additional examples.

Get the latest posts in your inbox

Subscribe to our blog and get the latest posts delivered to your inbox.

View all blogs
)\n for emails in iterator:\n yield pa.array([bool(pattern.match(e)) for e in emails.to_pylist()])Iterator of Multiple Arrays to Iterator of Arrays: receiving an iterator of a tuple of multiple pyarrow.Array and returning an iterator of pyarrow.Array.python@arrow_udf(\"long\")\ndef multiply(iterator: Iterator[Tuple[pa.Array, pa.Array]]) -> Iterator[pa.Array]:\n for v1, v2 in iterator:\n yield pa.compute.multiply(v1, v2)Arrow Aggregate FunctionsArrow Aggregate Functions take one or more pyarrow.Array inputs and return a scalar value, reducing a group of rows into a single result. They are the Arrow equivalent of grouped aggregate Pandas UDFs and are used with groupBy().agg() or Window operations. Similar to scalar functions, aggregate functions also support three input modes. Arrays to Scalar: receiving pyarrow.Array and returning a scalar value. python@arrow_udf(\"struct<m1: double, m2: double>\")\ndef min_max_udf(v: pa.Array) -> pa.Scalar:\n m1 = pa.compute.min(v)\n m2 = pa.compute.max(v)\n t = pa.struct([pa.field(\"m1\", pa.float64()), pa.field(\"m2\", pa.float64())])\n return pa.scalar(value={\"m1\": m1.as_py(), \"m2\": m2.as_py()}, type=t)Iterator of Arrays to Scalar: receiving an iterator of pyarrow.Array and returning a scalar value. This is useful for processing large volumes of data in aggregation-style operations.python@arrow_udf(\"double\")\ndef streaming_mean(iterator: Iterator[pa.Array]) -> float:\n total_sum = 0.0\n total_count = 0\n for batch in iterator:\n total_sum += pa.compute.sum(batch).as_py()\n total_count += len(batch)\n return total_sum / total_count if total_count > 0 else 0.0Iterator of Multiple Arrays to Scalar: receiving an iterator of a tuple of multiple pyarrow.Array and returning a scalar value. More complex aggregations can be defined.python@arrow_udf(\"double\")\ndef weighted_mean(iterator: Iterator[Tuple[pa.Array, pa.Array]]) -> float:\n weighted_sum = 0.0\n total_weight = 0.0\n for values, weights in iterator:\n weighted_sum += pa.compute.sum(pa.compute.multiply(values, weights)).as_py()\n total_weight += pa.compute.sum(weights).as_py()\n return weighted_sum / total_weight if total_weight > 0 else 0.0Arrow Table FunctionsArrow Table Functions, also known as Arrow UDTFs (User-Defined Table Functions), accept a pyarrow.RecordBatch or multiple pa.Array as input and produce a pyarrow.Table as output. This represents the predominant pattern for table-in, table-out transformations implemented in Python utilizing columnar execution. Arrow UDTFs possess the capability to:Return multiple columnsProduce zero, one, or multiple rowsExecute vectorized table transformations employing Arrow compute kernelsConsequently, they are optimally suited for operations such as filtering, row expansion, data restructuring, and the generation of derived columns.The arrow_udtf interface is designed for simplicity, employing a decorator syntax where you define the return type using a DDL-formatted string. In this setup, the eval method takes PyArrow objects as input and is expected to yield PyArrow Tables or RecordBatches. The interface accommodates two input modes. When processing table arguments, the eval method is provided with a pa.RecordBatch object that encapsulates all columns from the input table:python@arrow_udtf(returnType=\"x int, y int\")\nclass ProcessBatch:\n def eval(self, batch: pa.RecordBatch):\n x_array = batch.column('x')\n y_array = batch.column('y')\n result = pa.table({\n 'x': pc.multiply(x_array, 2),\n 'y': pc.add(y_array, 10)\n })\n yield resultFor scalar arguments, the method receives pa.Array objects, one for each scalar input:python@arrow_udtf(returnType=\"x int, y int\")\nclass ProcessArrays:\n def eval(self, x: pa.Array, y: pa.Array):\n result = pa.table({\n 'x': x,\n 'y': pc.multiply(y, 2)\n })\n yield resultHere is another example:python@arrow_udtf(returnType=\"sensor_id string, temp_f double, status string\")\nclass ProcessReadings:\n def eval(self, batch: pa.RecordBatch, threshold: pa.Array):\n min_temp = threshold[0].as_py()\n \n # Vectorized filter: keep readings above threshold\n mask = pc.greater(batch.column(\"temp_c\"), min_temp)\n filtered = pa.table(batch).filter(mask)\n \n # Vectorized transform: Celsius to Fahrenheit\n temp_f = pc.add(pc.multiply(filtered.column(\"temp_c\"), 1.8), 32)\n \n # Vectorized conditional: assign status based on temperature\n status = pc.if_else(\n pc.greater(filtered.column(\"temp_c\"), 35),\n pa.scalar(\"CRITICAL\"),\n pa.scalar(\"WARNING\")\n )\n \n yield pa.table({\n \"sensor_id\": filtered.column(\"sensor_id\"),\n \"temp_f\": temp_f,\n \"status\": status\n })This UDTF can work in two distinct ways:Python Usage:pythonfrom pyspark.sql import functions as F\n\n# Generate sample data\ndf = spark.createDataFrame([\n (\"sensor_1\", 30.0),\n (\"sensor_2\", 36.0),\n (\"sensor_3\", 25.0),\n], [\"sensor_id\", \"temp_c\"])\n\n# Invoke the UDTF within Python\nresult = ProcessReadings(df.asTable(), F.lit(28.0))\n\n\nresult.show()\n# +----------+-------+--------+\n# |sensor_id |temp_f |status |\n# +----------+-------+--------+\n# |sensor_1 |86.0 |WARNING |\n# |sensor_2 |96.8 |CRITICAL|\n# +----------+-------+--------+SQL Usage:python# Register the UDTF for SQL environment access\nspark.udtf.register(\"process_readings\", ProcessReadings)\n\n# Establish a temporary table or view\ndf.createOrReplaceTempView(\"sensor_data\")\n\n# Execute the UDTF in SQL\nspark.sql(\"\"\"\n SELECT *\n FROM process_readings(\n TABLE(SELECT sensor_id, temp_c FROM sensor_data),\n 28.0\n )\n\"\"\").show()\n# +----------+-------+--------+\n# |sensor_id |temp_f |status |\n# +----------+-------+--------+\n# |sensor_1 |86.0 |WARNING |\n# |sensor_2 |96.8 |CRITICAL|\n# +----------+-------+--------+DataFrame mapInArrow and applyInArrow SupportIn addition to User-Defined Functions (UDFs) and User-Defined Table Functions (UDTFs), PySpark furnishes Arrow Function APIs that facilitate the direct application of Python native functions to Arrow data at the DataFrame level. These APIs operate analogously to their Pandas counterparts (mapInPandas, applyInPandas) but utilize pyarrow.RecordBatch and pyarrow.Table instead of Pandas DataFrames, thereby circumventing the conversion overhead between Pandas and Arrow formats.Map. DataFrame.mapInArrow transforms an iterator of pyarrow.RecordBatch into another iterator of pyarrow.RecordBatch, enabling row-level operations such as filtering, transformation, or expansion.pythonimport pyarrow as pa\n\ndf = spark.createDataFrame([(1, 21), (2, 30)], (\"id\", \"age\"))\n\ndef filter_func(iterator):\n for batch in iterator:\n yield batch.filter(pa.compute.field(\"id\") == 1)\n\ndf.mapInArrow(filter_func, df.schema).show()\n# +---+---+\n# | id|age|\n# +---+---+\n# | 1| 21|\n# +---+---+Grouped Map. groupBy().applyInArrow() applies a specified function to each group, accepting and returning a pyarrow.Table. This functionality proves beneficial for per-group transformations, such as data normalization.pythonimport pyarrow as pa\nimport pyarrow.compute as pc\n\ndf = spark.createDataFrame(\n [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)], (\"id\", \"v\"))\n\ndef normalize(table):\n v = table.column(\"v\")\n norm = pc.divide(pc.subtract(v, pc.mean(v)), pc.stddev(v, ddof=1))\n return table.set_column(1, \"v\", norm)\n\ndf.groupby(\"id\").applyInArrow(normalize, schema=\"id long, v double\").show()\n# +---+-------------------+\n# | id| v|\n# +---+-------------------+\n# | 1|-0.7071067811865...|\n# | 1| 0.7071067811865...|\n# | 2|-0.8320502943378...|\n# | 2|-0.2773500981126...|\n# | 2| 1.1094003924504...|\n# +---+-------------------+Co-grouped Map. cogroup().applyInArrow() permits the cogrouping of two DataFrames based on a shared key, subsequently applying a function to each cogroup. The function receives two pyarrow.Table inputs and is expected to return a single pyarrow.Table.pythonimport pyarrow as pa\n\ndf1 = spark.createDataFrame(\n [(1, 1.0), (2, 2.0), (1, 3.0), (2, 4.0)], (\"id\", \"v1\"))\ndf2 = spark.createDataFrame([(1, \"x\"), (2, \"y\")], (\"id\", \"v2\"))\n\ndef summarize(l, r):\n return pa.Table.from_pydict({\n \"left\": [l.num_rows],\n \"right\": [r.num_rows]\n })\n\ndf1.groupby(\"id\").cogroup(df2.groupby(\"id\")).applyInArrow(\n summarize, schema=\"left long, right long\").show()\n# +----+-----+\n# |left|right|\n# +----+-----+\n# | 2| 1|\n# | 2| 1|\n# +----+-----+PerformanceBy removing the expensive Pandas/Arrow data conversion, Arrow UDFs generally execute faster than Pandas UDFs, with less memory usage. Let\u2019s compare the two simple UDFs:python@pandas_udf(\"long\")\ndef multiply_pandas_func(a: pd.Series, b: pd.Series) -> pd.Series:\n return a * b\n\n@arrow_udf(\"long\")\ndef multiply_arrow_func(a: pa.Array, b: pa.Array) -> pa.Array:\n return pa.compute.multiply(a, b)The Arrow UDF is ~10% faster than the Pandas UDF, and the memory profiler shows that ~40% memory is saved in the execution.python============================================================\nProfile of UDF<id=2>\n============================================================\nFilename: <ipython-input-1-af507ad7b0d2>\n\nLine # Mem usage Increment Occurrences Line Contents\n=============================================================\n 12 135.8 MiB 135.8 MiB 1 @pandas_udf(\"long\")\n 13 def multiply_pandas_func(a: pd.Series, b: pd.Series) -> pd.Series:\n 14 143.5 MiB 7.8 MiB 1 return a * b\n\n\n============================================================\nProfile of UDF<id=2>\n============================================================\nFilename: <ipython-input-1-0e8713529486>\n\nLine # Mem usage Increment Occurrences Line Contents\n=============================================================\n 11 71.7 MiB 71.7 MiB 1 @arrow_udf(\"long\")\n 12 def multiply_arrow_func(a: pa.Array, b: pa.Array) -> pa.Array:\n 13 79.9 MiB 8.2 MiB 1 return pa.compute.multiply(a, b)ConclusionDatabricks Runtime 18.0 introduces Native Arrow UDFs, offering a faster, leaner alternative to Pandas UDFs for performant Python UDF execution in PySpark. By operating directly on Arrow data and eliminating the Pandas/Arrow conversion overhead, Arrow UDFs deliver ~10% faster execution, ~40% less memory usage, and better support for complex datatypes -- all with a familiar, intuitive decorator syntax.Ready to explore more? Try out Native Arrow UDFs today on Databricks as part of Databricks Runtime 18.0. To get started, simply replace your existing Pandas UDFs with Arrow UDFs. In most cases, it only takes a few lines of change to unlock immediate performance gains. See the Arrow UDF documentation and Arrow UDTF documentation for the full API reference and additional examples.", "description": "Discover how to write more performant UDFs with native pyarrow support.", "name": "Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs", "mainEntityOfPage": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs", "headline": "Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs", "dateModified": "05/22/2026T00:00:00-08:00", "image": [{"@type": "ImageObject", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#BlogPosting_image_ImageObject", "url": "https://www.databricks.com/sites/default/files/2026-05/2026-02-blog-introducing-arrow-udfs-in-pyspark-inline-960x502.4.png"}], "datePublished": " 05/20/2026T00:00:00-08:00", "inLanguage": "en-US", "mentions": [{"@type": "BreadcrumbList", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#BlogPosting_mentions_BreadcrumbList", "itemListElement": [{"@type": "ListItem", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight20260508174503550-32645_0_BlogPosting_mentions_BreadcrumbList_itemListElement_ListItem", "name": "All blogs", "item": "https://www.databricks.com/blog", "position": 1}, {"@type": "ListItem", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight20260508174503550-32645_1_BlogPosting_mentions_BreadcrumbList_itemListElement_ListItem", "name": "Engineering", "item": "https://www.databricks.com/blog/category/engineering", "position": 2}]}, {"name": "Python", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_python_9ccca1ec3e5c4f56bb4441c4a6310e73b636329f3426d369a4b44d5ead8fdc42", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/05z1_", "http://www.wikidata.org/entity/Q28865", "https://en.wikipedia.org/wiki/Python_(programming_language)"]}, {"name": "Universal Disk Format", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_universaldiskformat_b725ccaa9e7dc56cca06f63ed569331f7c60555e552d09a063ac39da5ce21f68", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0d5x3", "http://www.wikidata.org/entity/Q853645", "https://en.wikipedia.org/wiki/Universal_Disk_Format"]}, {"name": "execution", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_execution_420ccf06eaa11b3985eae8084d4e698dfa96d5660f24f7dbf0a6a343690308b4", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q1077724"}, {"name": "Apache Spark", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_apachespark_5c8dd45dbca4e0e307dfd60fd118717da7b8685e7a2e0f8c1c76c42d24a27cf6", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0ndhxqz", "http://www.wikidata.org/entity/Q7573619", "https://en.wikipedia.org/wiki/Apache_Spark"]}, {"name": "native", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_native_97ba8527ef17f94987e6803a88c2f9b84dc1c18560fbc995918c8c56ff16b1b1", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/02s6ql", "http://www.wikidata.org/entity/Q2187908", "https://en.wikipedia.org/wiki/Native_(computing)"]}, {"name": "data definition language", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_datadefinitionlanguage_7da80b099ad3226ffb2f7ee8c8ec22a7dc933293850ddf7504fa27405d147251", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q1431648"}, {"name": "Databricks", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWorkOrganization_databricks_c3569f7e9ee8cf86b2e95b355447ce0044e7844d96cf987d3bbe3659b15412b2", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0120wgnc", "http://www.wikidata.org/entity/Q18350420", "https://en.wikipedia.org/wiki/Databricks"]}, {"name": "NULL", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_null_26868ea02dcb683db45b52d8edce2a7fb1f72b9096401e1590bff56bfd4fbbee", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q371029"}], "articleSection": "Login\nOpen Source", "author": [{"@type": "Person", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight-20250224200644452_0_BlogPosting_author_Person", "url": "https://www.databricks.com/blog/author/ruifeng-zheng", "name": "Ruifeng Zheng"}, {"@type": "Person", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight-20250224200644452_1_BlogPosting_author_Person", "url": "https://www.databricks.com/blog/author/yicong-huang", "name": "Yicong Huang"}]}, {"@context": "http://schema.org", "@type": "Organization", "description": "The Databricks Platform is the world\u2019s first data intelligence platform powered by generative AI. Infuse AI into every facet of your business.", "name": "Databricks", "disambiguatingDescription": "Your data. Your AI. Your future. Own them all on the new data intelligence platform", "sameAs": ["kg:/m/0120wgnc", "https://www.wikidata.org/wiki/Q18350420", "https://en.wikipedia.org/wiki/Databricks", "https://twitter.com/databricks", "https://www.databricks.com/feed", "https://www.youtube.com/c/Databricks", "https://www.facebook.com/pages/Databricks/560203607379694", "https://www.linkedin.com/company/databricks", "https://www.glassdoor.com/Overview/Working-at-Databricks-EI_IE954734.11,21.htm"], "telephone": "+1-866-330-0121", "areaServed": "http://www.wikidata.org/entity/Q13780930", "legalName": "Databricks Inc.", "knowsLanguage": "en-US", "url": "https://www.databricks.com/", "additionalType": "https://www.wikidata.org/wiki/Q110029326", "logo": {"@type": "ImageObject", "width": "127", "height": "20", "url": "https://www.databricks.com/en-website-assets/static/8ed15a13c1511a75a4855999a2011c5c/f2f26/databricks-default.webp", "@id": "https://www.databricks.com/en-website-assets/static/8ed15a13c1511a75a4855999a2011c5c/f2f26/databricks-default.webp"}, "contactPoint": {"@type": "ContactPoint", "contactOption": "TollFree", "availableLanguage": "en-US", "contactType": "Contact", "telephone": "+1-866-330-0121", "name": "Databricks Contact", "@id": "https://www.databricks.com/#ContactPoint"}, "address": {"@type": "PostalAddress", "streetAddress": "160 Spear Street", "postalCode": "94105", "addressRegion": ["https://www.wikidata.org/wiki/Q99", "California"], "addressLocality": ["https://www.wikidata.org/wiki/Q62", "San Francisco"], "addressCountry": "http://www.wikidata.org/entity/Q30", "name": "Databricks Address", "@id": "https://www.databricks.com/#PostalAddress"}, "knowsAbout": [{"@type": "Thing", "sameAs": ["kg:/g/11khkg2rwf", "https://www.wikidata.org/wiki/Q117246174", "https://en.wikipedia.org/wiki/Generative_artificial_intelligence"], "name": "Generative AI", "@id": "https://www.databricks.com/#Thing"}, {"@type": "Thing", "sameAs": ["kg:/m/038_34", "https://en.wikipedia.org/wiki/Data_management", "https://www.wikidata.org/wiki/Q1149776"], "name": "data management", "@id": "https://www.databricks.com/#Thing1"}, {"@type": "Thing", "sameAs": ["kg:/m/0136zzks", "https://en.wikipedia.org/wiki/Data_lake", "https://www.wikidata.org/wiki/Q20707560"], "name": "data lake", "@id": "https://www.databricks.com/#Thing2"}, {"@type": "Thing", "sameAs": ["kg:/m/0h7m73m", "https://www.wikidata.org/wiki/Q2499178", "https://en.wikipedia.org/wiki/Cloud_database"], "name": "cloud database", "@id": "https://www.databricks.com/#Thing3"}, {"@type": "Thing", "sameAs": ["kg:/m/0mkz", "https://en.wikipedia.org/wiki/Artificial_intelligence", "https://www.wikidata.org/wiki/Q11660"], "name": "artificial intelligence", "@id": "https://www.databricks.com/#Thing4"}, {"@type": "Thing", "sameAs": ["kg:/m/01hyh", "https://www.wikidata.org/wiki/Q2539", "https://en.wikipedia.org/wiki/Machine_learning"], "name": "machine learning", "@id": "https://www.databricks.com/#Thing5"}], "image": {"@type": "ImageObject", "width": "1200", "height": "628", "url": "https://www.databricks.com/sites/default/files/2023-11/databricks-og-universal.png", "@id": "https://www.databricks.com/sites/default/files/2023-11/databricks-og-universal.png"}, "@id": "https://www.databricks.com/#Organization"}]
Skip to main content
Open Source

Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs

Define more performant UDFs with ease.

by Ruifeng Zheng and Yicong Huang

  • We introduce native Arrow UDFs, which operate directly on Arrow data, eliminating the Pandas/Arrow conversion overhead in Pandas UDFs for faster execution and lower memory usage.
  • We also describe Arrow UDF types for scalar and aggregation use cases, and Arrow UDTFs for table-in, table-out transformations, with code examples in both Python and SQL.
  • Benchmarks show Arrow UDFs are ~10% faster and use ~40% less memory than Pandas UDFs, with better support for complex datatypes.

Introduction

Python User-Defined Functions (UDFs) are an essential extensibility mechanism but have traditionally suffered from high overhead due to row-based execution. In Apache Spark™, Pandas UDFs addressed part of this problem by introducing Arrow-based serialization and batch processing, significantly improving throughput compared to scalar Python UDFs.

However, Pandas UDFs still have fundamental limitations:

  • The Pandas/Arrow data conversion introduces additional data copies. Zero-copy approaches are only possible in certain narrow cases. For example, columns with NULL values will trigger deep copies.
  • Complex datatypes are not supported well. For example, nested StructType instances are not supported for the output type with aggregation use cases.
Data flows of Pandas UDF Execution in Apache Spark™

By dropping the Pandas/Arrow data conversion, the Arrow UDFs execute faster than Pandas UDFs, consume less memory, and provide better datatype support.

Native Arrow UDFs

We’re thrilled to introduce Native Arrow UDFs starting with Databricks Runtime 18.0 (release notes), an exciting leap forward for performant UDF execution.

Native Arrow UDFs operate directly on Arrow data without converting inputs into Pandas or NumPy objects. This preserves the columnar layout end-to-end, avoids unnecessary data copies, and lets UDFs use vectorized processing by leveraging Arrow’s native compute and memory model.

To define an Arrow UDF, users are able to use a new python decorator @arrow_udf, with specified return type and optional evaluation type. For instance:

Users can also define it with existing decorator @udf with complete type hints. For instance:

Note: The function definition should include type hints for all of the arguments and the return value.
This design aligns with the interfaces of scalar Python UDFs, providing a consistent and intuitive experience for users already familiar with scalar Python UDFs.

The following demonstrates how to use the Arrow UDF:

Python Usage:

SQL Usage:

We provide support for variants of Arrow UDF interfaces. Including Scalar Functions, Aggregate Functions and Table Functions. In the data frame API we also provide mapInArrow and applyInArrow to use Arrow UDFs. We will next introduce them one by one. 

Arrow Scalar Functions

Arrow Scalar Functions perform row-wise transformations. They are the Arrow equivalent of scalar Pandas UDFs and can be used anywhere a column expression is expected, such as df.select() or df.withColumn(). Three input modes are supported: direct, iterator, and iterator of multiple arrays. The iterator variants are useful when the UDF requires expensive one-time initialization (e.g.,  loading a model or compiling a regex pattern), as the setup cost is amortized across all batches. In all cases, the output row count must match the input row count.

  • Arrays to Array: receiving one or more pyarrow.Array and returning one pyarrow.Array. The input and output array must have the same number of values.
  • Iterator of Arrays to Iterator of Arrays: receiving an iterator of pyarrow.Array and returning an iterator of pyarrow.Array. This type is useful when the UDF execution requires expensive initialization. 
  • Iterator of Multiple Arrays to Iterator of Arrays: receiving an iterator of a tuple of multiple pyarrow.Array and returning an iterator of pyarrow.Array.

Arrow Aggregate Functions

Arrow Aggregate Functions take one or more pyarrow.Array inputs and return a scalar value, reducing a group of rows into a single result. They are the Arrow equivalent of grouped aggregate Pandas UDFs and are used with groupBy().agg() or Window operations. Similar to scalar functions, aggregate functions also support three input modes. 

Arrays to Scalar: receiving pyarrow.Array and returning a scalar value. 

  • Iterator of Arrays to Scalar: receiving an iterator of pyarrow.Array and returning a scalar value. This is useful for processing large volumes of data in aggregation-style operations.

Iterator of Multiple Arrays to Scalar: receiving an iterator of a tuple of multiple pyarrow.Array and returning a scalar value. More complex aggregations can be defined.

Arrow Table Functions

Arrow Table Functions, also known as Arrow UDTFs (User-Defined Table Functions), accept a pyarrow.RecordBatch or multiple pa.Array as input and produce a pyarrow.Table as output. This represents the predominant pattern for table-in, table-out transformations implemented in Python utilizing columnar execution. Arrow UDTFs possess the capability to:

  • Return multiple columns
  • Produce zero, one, or multiple rows
  • Execute vectorized table transformations employing Arrow compute kernels

Consequently, they are optimally suited for operations such as filtering, row expansion, data restructuring, and the generation of derived columns.

The arrow_udtf interface is designed for simplicity, employing a decorator syntax where you define the return type using a DDL-formatted string. In this setup, the eval method takes PyArrow objects as input and is expected to yield PyArrow Tables or RecordBatches. The interface accommodates two input modes. When processing table arguments, the eval method is provided with a pa.RecordBatch object that encapsulates all columns from the input table:

For scalar arguments, the method receives pa.Array objects, one for each scalar input:

Here is another example:

This UDTF can work in two distinct ways:

Python Usage:

SQL Usage:

DataFrame mapInArrow and applyInArrow Support

In addition to User-Defined Functions (UDFs) and User-Defined Table Functions (UDTFs), PySpark furnishes Arrow Function APIs that facilitate the direct application of Python native functions to Arrow data at the DataFrame level. These APIs operate analogously to their Pandas counterparts (mapInPandas, applyInPandas) but utilize pyarrow.RecordBatch and pyarrow.Table instead of Pandas DataFrames, thereby circumventing the conversion overhead between Pandas and Arrow formats.

  • Map. DataFrame.mapInArrow transforms an iterator of pyarrow.RecordBatch into another iterator of pyarrow.RecordBatch, enabling row-level operations such as filtering, transformation, or expansion.
  • Grouped Map. groupBy().applyInArrow() applies a specified function to each group, accepting and returning a pyarrow.Table. This functionality proves beneficial for per-group transformations, such as data normalization.
  • Co-grouped Map. cogroup().applyInArrow() permits the cogrouping of two DataFrames based on a shared key, subsequently applying a function to each cogroup. The function receives two pyarrow.Table inputs and is expected to return a single pyarrow.Table.

Performance

By removing the expensive Pandas/Arrow data conversion, Arrow UDFs generally execute faster than Pandas UDFs, with less memory usage. Let’s compare the two simple UDFs:

The Arrow UDF is ~10% faster than the Pandas UDF, and the memory profiler shows that ~40% memory is saved in the execution.

Conclusion

Databricks Runtime 18.0 introduces Native Arrow UDFs, offering a faster, leaner alternative to Pandas UDFs for performant Python UDF execution in PySpark. By operating directly on Arrow data and eliminating the Pandas/Arrow conversion overhead, Arrow UDFs deliver ~10% faster execution, ~40% less memory usage, and better support for complex datatypes -- all with a familiar, intuitive decorator syntax.

Ready to explore more? Try out Native Arrow UDFs today on Databricks as part of Databricks Runtime 18.0. To get started, simply replace your existing Pandas UDFs with Arrow UDFs. In most cases, it only takes a few lines of change to unlock immediate performance gains. See the Arrow UDF documentation and Arrow UDTF documentation for the full API reference and additional examples.

Get the latest posts in your inbox

Subscribe to our blog and get the latest posts delivered to your inbox.

View all blogs
)\n for emails in iterator:\n yield pa.array([bool(pattern.match(e)) for e in emails.to_pylist()])Iterator of Multiple Arrays to Iterator of Arrays: receiving an iterator of a tuple of multiple pyarrow.Array and returning an iterator of pyarrow.Array.python@arrow_udf(\"long\")\ndef multiply(iterator: Iterator[Tuple[pa.Array, pa.Array]]) -> Iterator[pa.Array]:\n for v1, v2 in iterator:\n yield pa.compute.multiply(v1, v2)Arrow Aggregate FunctionsArrow Aggregate Functions take one or more pyarrow.Array inputs and return a scalar value, reducing a group of rows into a single result. They are the Arrow equivalent of grouped aggregate Pandas UDFs and are used with groupBy().agg() or Window operations. Similar to scalar functions, aggregate functions also support three input modes. Arrays to Scalar: receiving pyarrow.Array and returning a scalar value. python@arrow_udf(\"struct<m1: double, m2: double>\")\ndef min_max_udf(v: pa.Array) -> pa.Scalar:\n m1 = pa.compute.min(v)\n m2 = pa.compute.max(v)\n t = pa.struct([pa.field(\"m1\", pa.float64()), pa.field(\"m2\", pa.float64())])\n return pa.scalar(value={\"m1\": m1.as_py(), \"m2\": m2.as_py()}, type=t)Iterator of Arrays to Scalar: receiving an iterator of pyarrow.Array and returning a scalar value. This is useful for processing large volumes of data in aggregation-style operations.python@arrow_udf(\"double\")\ndef streaming_mean(iterator: Iterator[pa.Array]) -> float:\n total_sum = 0.0\n total_count = 0\n for batch in iterator:\n total_sum += pa.compute.sum(batch).as_py()\n total_count += len(batch)\n return total_sum / total_count if total_count > 0 else 0.0Iterator of Multiple Arrays to Scalar: receiving an iterator of a tuple of multiple pyarrow.Array and returning a scalar value. More complex aggregations can be defined.python@arrow_udf(\"double\")\ndef weighted_mean(iterator: Iterator[Tuple[pa.Array, pa.Array]]) -> float:\n weighted_sum = 0.0\n total_weight = 0.0\n for values, weights in iterator:\n weighted_sum += pa.compute.sum(pa.compute.multiply(values, weights)).as_py()\n total_weight += pa.compute.sum(weights).as_py()\n return weighted_sum / total_weight if total_weight > 0 else 0.0Arrow Table FunctionsArrow Table Functions, also known as Arrow UDTFs (User-Defined Table Functions), accept a pyarrow.RecordBatch or multiple pa.Array as input and produce a pyarrow.Table as output. This represents the predominant pattern for table-in, table-out transformations implemented in Python utilizing columnar execution. Arrow UDTFs possess the capability to:Return multiple columnsProduce zero, one, or multiple rowsExecute vectorized table transformations employing Arrow compute kernelsConsequently, they are optimally suited for operations such as filtering, row expansion, data restructuring, and the generation of derived columns.The arrow_udtf interface is designed for simplicity, employing a decorator syntax where you define the return type using a DDL-formatted string. In this setup, the eval method takes PyArrow objects as input and is expected to yield PyArrow Tables or RecordBatches. The interface accommodates two input modes. When processing table arguments, the eval method is provided with a pa.RecordBatch object that encapsulates all columns from the input table:python@arrow_udtf(returnType=\"x int, y int\")\nclass ProcessBatch:\n def eval(self, batch: pa.RecordBatch):\n x_array = batch.column('x')\n y_array = batch.column('y')\n result = pa.table({\n 'x': pc.multiply(x_array, 2),\n 'y': pc.add(y_array, 10)\n })\n yield resultFor scalar arguments, the method receives pa.Array objects, one for each scalar input:python@arrow_udtf(returnType=\"x int, y int\")\nclass ProcessArrays:\n def eval(self, x: pa.Array, y: pa.Array):\n result = pa.table({\n 'x': x,\n 'y': pc.multiply(y, 2)\n })\n yield resultHere is another example:python@arrow_udtf(returnType=\"sensor_id string, temp_f double, status string\")\nclass ProcessReadings:\n def eval(self, batch: pa.RecordBatch, threshold: pa.Array):\n min_temp = threshold[0].as_py()\n \n # Vectorized filter: keep readings above threshold\n mask = pc.greater(batch.column(\"temp_c\"), min_temp)\n filtered = pa.table(batch).filter(mask)\n \n # Vectorized transform: Celsius to Fahrenheit\n temp_f = pc.add(pc.multiply(filtered.column(\"temp_c\"), 1.8), 32)\n \n # Vectorized conditional: assign status based on temperature\n status = pc.if_else(\n pc.greater(filtered.column(\"temp_c\"), 35),\n pa.scalar(\"CRITICAL\"),\n pa.scalar(\"WARNING\")\n )\n \n yield pa.table({\n \"sensor_id\": filtered.column(\"sensor_id\"),\n \"temp_f\": temp_f,\n \"status\": status\n })This UDTF can work in two distinct ways:Python Usage:pythonfrom pyspark.sql import functions as F\n\n# Generate sample data\ndf = spark.createDataFrame([\n (\"sensor_1\", 30.0),\n (\"sensor_2\", 36.0),\n (\"sensor_3\", 25.0),\n], [\"sensor_id\", \"temp_c\"])\n\n# Invoke the UDTF within Python\nresult = ProcessReadings(df.asTable(), F.lit(28.0))\n\n\nresult.show()\n# +----------+-------+--------+\n# |sensor_id |temp_f |status |\n# +----------+-------+--------+\n# |sensor_1 |86.0 |WARNING |\n# |sensor_2 |96.8 |CRITICAL|\n# +----------+-------+--------+SQL Usage:python# Register the UDTF for SQL environment access\nspark.udtf.register(\"process_readings\", ProcessReadings)\n\n# Establish a temporary table or view\ndf.createOrReplaceTempView(\"sensor_data\")\n\n# Execute the UDTF in SQL\nspark.sql(\"\"\"\n SELECT *\n FROM process_readings(\n TABLE(SELECT sensor_id, temp_c FROM sensor_data),\n 28.0\n )\n\"\"\").show()\n# +----------+-------+--------+\n# |sensor_id |temp_f |status |\n# +----------+-------+--------+\n# |sensor_1 |86.0 |WARNING |\n# |sensor_2 |96.8 |CRITICAL|\n# +----------+-------+--------+DataFrame mapInArrow and applyInArrow SupportIn addition to User-Defined Functions (UDFs) and User-Defined Table Functions (UDTFs), PySpark furnishes Arrow Function APIs that facilitate the direct application of Python native functions to Arrow data at the DataFrame level. These APIs operate analogously to their Pandas counterparts (mapInPandas, applyInPandas) but utilize pyarrow.RecordBatch and pyarrow.Table instead of Pandas DataFrames, thereby circumventing the conversion overhead between Pandas and Arrow formats.Map. DataFrame.mapInArrow transforms an iterator of pyarrow.RecordBatch into another iterator of pyarrow.RecordBatch, enabling row-level operations such as filtering, transformation, or expansion.pythonimport pyarrow as pa\n\ndf = spark.createDataFrame([(1, 21), (2, 30)], (\"id\", \"age\"))\n\ndef filter_func(iterator):\n for batch in iterator:\n yield batch.filter(pa.compute.field(\"id\") == 1)\n\ndf.mapInArrow(filter_func, df.schema).show()\n# +---+---+\n# | id|age|\n# +---+---+\n# | 1| 21|\n# +---+---+Grouped Map. groupBy().applyInArrow() applies a specified function to each group, accepting and returning a pyarrow.Table. This functionality proves beneficial for per-group transformations, such as data normalization.pythonimport pyarrow as pa\nimport pyarrow.compute as pc\n\ndf = spark.createDataFrame(\n [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)], (\"id\", \"v\"))\n\ndef normalize(table):\n v = table.column(\"v\")\n norm = pc.divide(pc.subtract(v, pc.mean(v)), pc.stddev(v, ddof=1))\n return table.set_column(1, \"v\", norm)\n\ndf.groupby(\"id\").applyInArrow(normalize, schema=\"id long, v double\").show()\n# +---+-------------------+\n# | id| v|\n# +---+-------------------+\n# | 1|-0.7071067811865...|\n# | 1| 0.7071067811865...|\n# | 2|-0.8320502943378...|\n# | 2|-0.2773500981126...|\n# | 2| 1.1094003924504...|\n# +---+-------------------+Co-grouped Map. cogroup().applyInArrow() permits the cogrouping of two DataFrames based on a shared key, subsequently applying a function to each cogroup. The function receives two pyarrow.Table inputs and is expected to return a single pyarrow.Table.pythonimport pyarrow as pa\n\ndf1 = spark.createDataFrame(\n [(1, 1.0), (2, 2.0), (1, 3.0), (2, 4.0)], (\"id\", \"v1\"))\ndf2 = spark.createDataFrame([(1, \"x\"), (2, \"y\")], (\"id\", \"v2\"))\n\ndef summarize(l, r):\n return pa.Table.from_pydict({\n \"left\": [l.num_rows],\n \"right\": [r.num_rows]\n })\n\ndf1.groupby(\"id\").cogroup(df2.groupby(\"id\")).applyInArrow(\n summarize, schema=\"left long, right long\").show()\n# +----+-----+\n# |left|right|\n# +----+-----+\n# | 2| 1|\n# | 2| 1|\n# +----+-----+PerformanceBy removing the expensive Pandas/Arrow data conversion, Arrow UDFs generally execute faster than Pandas UDFs, with less memory usage. Let\u2019s compare the two simple UDFs:python@pandas_udf(\"long\")\ndef multiply_pandas_func(a: pd.Series, b: pd.Series) -> pd.Series:\n return a * b\n\n@arrow_udf(\"long\")\ndef multiply_arrow_func(a: pa.Array, b: pa.Array) -> pa.Array:\n return pa.compute.multiply(a, b)The Arrow UDF is ~10% faster than the Pandas UDF, and the memory profiler shows that ~40% memory is saved in the execution.python============================================================\nProfile of UDF<id=2>\n============================================================\nFilename: <ipython-input-1-af507ad7b0d2>\n\nLine # Mem usage Increment Occurrences Line Contents\n=============================================================\n 12 135.8 MiB 135.8 MiB 1 @pandas_udf(\"long\")\n 13 def multiply_pandas_func(a: pd.Series, b: pd.Series) -> pd.Series:\n 14 143.5 MiB 7.8 MiB 1 return a * b\n\n\n============================================================\nProfile of UDF<id=2>\n============================================================\nFilename: <ipython-input-1-0e8713529486>\n\nLine # Mem usage Increment Occurrences Line Contents\n=============================================================\n 11 71.7 MiB 71.7 MiB 1 @arrow_udf(\"long\")\n 12 def multiply_arrow_func(a: pa.Array, b: pa.Array) -> pa.Array:\n 13 79.9 MiB 8.2 MiB 1 return pa.compute.multiply(a, b)ConclusionDatabricks Runtime 18.0 introduces Native Arrow UDFs, offering a faster, leaner alternative to Pandas UDFs for performant Python UDF execution in PySpark. By operating directly on Arrow data and eliminating the Pandas/Arrow conversion overhead, Arrow UDFs deliver ~10% faster execution, ~40% less memory usage, and better support for complex datatypes -- all with a familiar, intuitive decorator syntax.Ready to explore more? Try out Native Arrow UDFs today on Databricks as part of Databricks Runtime 18.0. To get started, simply replace your existing Pandas UDFs with Arrow UDFs. In most cases, it only takes a few lines of change to unlock immediate performance gains. See the Arrow UDF documentation and Arrow UDTF documentation for the full API reference and additional examples.", "dateModified": "05/22/2026T00:00:00-08:00", "headline": "Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs", "image": [{"@type": "ImageObject", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#BlogPosting_image_ImageObject", "url": "https://www.databricks.com/sites/default/files/2026-05/2026-02-blog-introducing-arrow-udfs-in-pyspark-inline-960x502.4.png"}], "datePublished": " 05/20/2026T00:00:00-08:00", "mainEntityOfPage": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs", "inLanguage": "en-US", "name": "Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs", "articleSection": "Login\nOpen Source", "author": [{"@type": "Person", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight-20250224200644452_0_BlogPosting_author_Person", "url": "https://www.databricks.com/blog/author/ruifeng-zheng", "name": "Ruifeng Zheng"}, {"@type": "Person", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight-20250224200644452_1_BlogPosting_author_Person", "url": "https://www.databricks.com/blog/author/yicong-huang", "name": "Yicong Huang"}], "mentions": [{"@type": "BreadcrumbList", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#BlogPosting_mentions_BreadcrumbList", "itemListElement": [{"@type": "ListItem", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight20260508174503550-32645_0_BlogPosting_mentions_BreadcrumbList_itemListElement_ListItem", "name": "All blogs", "item": "https://www.databricks.com/blog", "position": 1}, {"@type": "ListItem", "@id": "https://www.databricks.com/blog/introducing-arrow-udfs-pyspark-faster-leaner-replacement-pandas-udfs#Highlight20260508174503550-32645_1_BlogPosting_mentions_BreadcrumbList_itemListElement_ListItem", "name": "Engineering", "item": "https://www.databricks.com/blog/category/engineering", "position": 2}]}, {"name": "Python", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_python_9ccca1ec3e5c4f56bb4441c4a6310e73b636329f3426d369a4b44d5ead8fdc42", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/05z1_", "http://www.wikidata.org/entity/Q28865", "https://en.wikipedia.org/wiki/Python_(programming_language)"]}, {"name": "Universal Disk Format", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_universaldiskformat_b725ccaa9e7dc56cca06f63ed569331f7c60555e552d09a063ac39da5ce21f68", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0d5x3", "http://www.wikidata.org/entity/Q853645", "https://en.wikipedia.org/wiki/Universal_Disk_Format"]}, {"name": "execution", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_execution_420ccf06eaa11b3985eae8084d4e698dfa96d5660f24f7dbf0a6a343690308b4", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q1077724"}, {"name": "Apache Spark", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_apachespark_5c8dd45dbca4e0e307dfd60fd118717da7b8685e7a2e0f8c1c76c42d24a27cf6", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0ndhxqz", "http://www.wikidata.org/entity/Q7573619", "https://en.wikipedia.org/wiki/Apache_Spark"]}, {"name": "native", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWork_native_97ba8527ef17f94987e6803a88c2f9b84dc1c18560fbc995918c8c56ff16b1b1", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/02s6ql", "http://www.wikidata.org/entity/Q2187908", "https://en.wikipedia.org/wiki/Native_(computing)"]}, {"name": "data definition language", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_datadefinitionlanguage_7da80b099ad3226ffb2f7ee8c8ec22a7dc933293850ddf7504fa27405d147251", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q1431648"}, {"name": "Databricks", "@id": "https://entity.schemaapp.com/DatabricksInc/CreativeWorkOrganization_databricks_c3569f7e9ee8cf86b2e95b355447ce0044e7844d96cf987d3bbe3659b15412b2", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": ["http://g.co/kg/m/0120wgnc", "http://www.wikidata.org/entity/Q18350420", "https://en.wikipedia.org/wiki/Databricks"]}, {"name": "NULL", "@id": "https://entity.schemaapp.com/DatabricksInc/Thing_null_26868ea02dcb683db45b52d8edce2a7fb1f72b9096401e1590bff56bfd4fbbee", "@type": "Thing", "@context": {"@vocab": "http://schema.org/"}, "sameAs": "http://www.wikidata.org/entity/Q371029"}]}, {"@context": "http://schema.org", "@type": "Organization", "description": "The Databricks Platform is the world\u2019s first data intelligence platform powered by generative AI. Infuse AI into every facet of your business.", "name": "Databricks", "disambiguatingDescription": "Your data. Your AI. Your future. Own them all on the new data intelligence platform", "sameAs": ["kg:/m/0120wgnc", "https://www.wikidata.org/wiki/Q18350420", "https://en.wikipedia.org/wiki/Databricks", "https://twitter.com/databricks", "https://www.databricks.com/feed", "https://www.youtube.com/c/Databricks", "https://www.facebook.com/pages/Databricks/560203607379694", "https://www.linkedin.com/company/databricks", "https://www.glassdoor.com/Overview/Working-at-Databricks-EI_IE954734.11,21.htm"], "telephone": "+1-866-330-0121", "areaServed": "http://www.wikidata.org/entity/Q13780930", "legalName": "Databricks Inc.", "knowsLanguage": "en-US", "url": "https://www.databricks.com/", "additionalType": "https://www.wikidata.org/wiki/Q110029326", "logo": {"@type": "ImageObject", "width": "127", "height": "20", "url": "https://www.databricks.com/en-website-assets/static/8ed15a13c1511a75a4855999a2011c5c/f2f26/databricks-default.webp", "@id": "https://www.databricks.com/en-website-assets/static/8ed15a13c1511a75a4855999a2011c5c/f2f26/databricks-default.webp"}, "contactPoint": {"@type": "ContactPoint", "contactOption": "TollFree", "availableLanguage": "en-US", "contactType": "Contact", "telephone": "+1-866-330-0121", "name": "Databricks Contact", "@id": "https://www.databricks.com/#ContactPoint"}, "address": {"@type": "PostalAddress", "streetAddress": "160 Spear Street", "postalCode": "94105", "addressRegion": ["https://www.wikidata.org/wiki/Q99", "California"], "addressLocality": ["https://www.wikidata.org/wiki/Q62", "San Francisco"], "addressCountry": "http://www.wikidata.org/entity/Q30", "name": "Databricks Address", "@id": "https://www.databricks.com/#PostalAddress"}, "knowsAbout": [{"@type": "Thing", "sameAs": ["kg:/g/11khkg2rwf", "https://www.wikidata.org/wiki/Q117246174", "https://en.wikipedia.org/wiki/Generative_artificial_intelligence"], "name": "Generative AI", "@id": "https://www.databricks.com/#Thing"}, {"@type": "Thing", "sameAs": ["kg:/m/038_34", "https://en.wikipedia.org/wiki/Data_management", "https://www.wikidata.org/wiki/Q1149776"], "name": "data management", "@id": "https://www.databricks.com/#Thing1"}, {"@type": "Thing", "sameAs": ["kg:/m/0136zzks", "https://en.wikipedia.org/wiki/Data_lake", "https://www.wikidata.org/wiki/Q20707560"], "name": "data lake", "@id": "https://www.databricks.com/#Thing2"}, {"@type": "Thing", "sameAs": ["kg:/m/0h7m73m", "https://www.wikidata.org/wiki/Q2499178", "https://en.wikipedia.org/wiki/Cloud_database"], "name": "cloud database", "@id": "https://www.databricks.com/#Thing3"}, {"@type": "Thing", "sameAs": ["kg:/m/0mkz", "https://en.wikipedia.org/wiki/Artificial_intelligence", "https://www.wikidata.org/wiki/Q11660"], "name": "artificial intelligence", "@id": "https://www.databricks.com/#Thing4"}, {"@type": "Thing", "sameAs": ["kg:/m/01hyh", "https://www.wikidata.org/wiki/Q2539", "https://en.wikipedia.org/wiki/Machine_learning"], "name": "machine learning", "@id": "https://www.databricks.com/#Thing5"}], "image": {"@type": "ImageObject", "width": "1200", "height": "628", "url": "https://www.databricks.com/sites/default/files/2023-11/databricks-og-universal.png", "@id": "https://www.databricks.com/sites/default/files/2023-11/databricks-og-universal.png"}, "@id": "https://www.databricks.com/#Organization"}]
Skip to main content
Open Source

Introducing Arrow UDFs in PySpark: A Faster, Leaner Replacement for Pandas UDFs

Define more performant UDFs with ease.

by Ruifeng Zheng and Yicong Huang

  • We introduce native Arrow UDFs, which operate directly on Arrow data, eliminating the Pandas/Arrow conversion overhead in Pandas UDFs for faster execution and lower memory usage.
  • We also describe Arrow UDF types for scalar and aggregation use cases, and Arrow UDTFs for table-in, table-out transformations, with code examples in both Python and SQL.
  • Benchmarks show Arrow UDFs are ~10% faster and use ~40% less memory than Pandas UDFs, with better support for complex datatypes.

Introduction

Python User-Defined Functions (UDFs) are an essential extensibility mechanism but have traditionally suffered from high overhead due to row-based execution. In Apache Spark™, Pandas UDFs addressed part of this problem by introducing Arrow-based serialization and batch processing, significantly improving throughput compared to scalar Python UDFs.

However, Pandas UDFs still have fundamental limitations:

  • The Pandas/Arrow data conversion introduces additional data copies. Zero-copy approaches are only possible in certain narrow cases. For example, columns with NULL values will trigger deep copies.
  • Complex datatypes are not supported well. For example, nested StructType instances are not supported for the output type with aggregation use cases.
Data flows of Pandas UDF Execution in Apache Spark™

By dropping the Pandas/Arrow data conversion, the Arrow UDFs execute faster than Pandas UDFs, consume less memory, and provide better datatype support.

Native Arrow UDFs

We’re thrilled to introduce Native Arrow UDFs starting with Databricks Runtime 18.0 (release notes), an exciting leap forward for performant UDF execution.

Native Arrow UDFs operate directly on Arrow data without converting inputs into Pandas or NumPy objects. This preserves the columnar layout end-to-end, avoids unnecessary data copies, and lets UDFs use vectorized processing by leveraging Arrow’s native compute and memory model.

To define an Arrow UDF, users are able to use a new python decorator @arrow_udf, with specified return type and optional evaluation type. For instance:

Users can also define it with existing decorator @udf with complete type hints. For instance:

Note: The function definition should include type hints for all of the arguments and the return value.
This design aligns with the interfaces of scalar Python UDFs, providing a consistent and intuitive experience for users already familiar with scalar Python UDFs.

The following demonstrates how to use the Arrow UDF:

Python Usage:

SQL Usage:

We provide support for variants of Arrow UDF interfaces. Including Scalar Functions, Aggregate Functions and Table Functions. In the data frame API we also provide mapInArrow and applyInArrow to use Arrow UDFs. We will next introduce them one by one. 

Arrow Scalar Functions

Arrow Scalar Functions perform row-wise transformations. They are the Arrow equivalent of scalar Pandas UDFs and can be used anywhere a column expression is expected, such as df.select() or df.withColumn(). Three input modes are supported: direct, iterator, and iterator of multiple arrays. The iterator variants are useful when the UDF requires expensive one-time initialization (e.g.,  loading a model or compiling a regex pattern), as the setup cost is amortized across all batches. In all cases, the output row count must match the input row count.

  • Arrays to Array: receiving one or more pyarrow.Array and returning one pyarrow.Array. The input and output array must have the same number of values.
  • Iterator of Arrays to Iterator of Arrays: receiving an iterator of pyarrow.Array and returning an iterator of pyarrow.Array. This type is useful when the UDF execution requires expensive initialization. 
  • Iterator of Multiple Arrays to Iterator of Arrays: receiving an iterator of a tuple of multiple pyarrow.Array and returning an iterator of pyarrow.Array.

Arrow Aggregate Functions

Arrow Aggregate Functions take one or more pyarrow.Array inputs and return a scalar value, reducing a group of rows into a single result. They are the Arrow equivalent of grouped aggregate Pandas UDFs and are used with groupBy().agg() or Window operations. Similar to scalar functions, aggregate functions also support three input modes. 

Arrays to Scalar: receiving pyarrow.Array and returning a scalar value. 

  • Iterator of Arrays to Scalar: receiving an iterator of pyarrow.Array and returning a scalar value. This is useful for processing large volumes of data in aggregation-style operations.

Iterator of Multiple Arrays to Scalar: receiving an iterator of a tuple of multiple pyarrow.Array and returning a scalar value. More complex aggregations can be defined.

Arrow Table Functions

Arrow Table Functions, also known as Arrow UDTFs (User-Defined Table Functions), accept a pyarrow.RecordBatch or multiple pa.Array as input and produce a pyarrow.Table as output. This represents the predominant pattern for table-in, table-out transformations implemented in Python utilizing columnar execution. Arrow UDTFs possess the capability to:

  • Return multiple columns
  • Produce zero, one, or multiple rows
  • Execute vectorized table transformations employing Arrow compute kernels

Consequently, they are optimally suited for operations such as filtering, row expansion, data restructuring, and the generation of derived columns.

The arrow_udtf interface is designed for simplicity, employing a decorator syntax where you define the return type using a DDL-formatted string. In this setup, the eval method takes PyArrow objects as input and is expected to yield PyArrow Tables or RecordBatches. The interface accommodates two input modes. When processing table arguments, the eval method is provided with a pa.RecordBatch object that encapsulates all columns from the input table:

For scalar arguments, the method receives pa.Array objects, one for each scalar input:

Here is another example:

This UDTF can work in two distinct ways:

Python Usage:

SQL Usage:

DataFrame mapInArrow and applyInArrow Support

In addition to User-Defined Functions (UDFs) and User-Defined Table Functions (UDTFs), PySpark furnishes Arrow Function APIs that facilitate the direct application of Python native functions to Arrow data at the DataFrame level. These APIs operate analogously to their Pandas counterparts (mapInPandas, applyInPandas) but utilize pyarrow.RecordBatch and pyarrow.Table instead of Pandas DataFrames, thereby circumventing the conversion overhead between Pandas and Arrow formats.

  • Map. DataFrame.mapInArrow transforms an iterator of pyarrow.RecordBatch into another iterator of pyarrow.RecordBatch, enabling row-level operations such as filtering, transformation, or expansion.
  • Grouped Map. groupBy().applyInArrow() applies a specified function to each group, accepting and returning a pyarrow.Table. This functionality proves beneficial for per-group transformations, such as data normalization.
  • Co-grouped Map. cogroup().applyInArrow() permits the cogrouping of two DataFrames based on a shared key, subsequently applying a function to each cogroup. The function receives two pyarrow.Table inputs and is expected to return a single pyarrow.Table.

Performance

By removing the expensive Pandas/Arrow data conversion, Arrow UDFs generally execute faster than Pandas UDFs, with less memory usage. Let’s compare the two simple UDFs:

The Arrow UDF is ~10% faster than the Pandas UDF, and the memory profiler shows that ~40% memory is saved in the execution.

Conclusion

Databricks Runtime 18.0 introduces Native Arrow UDFs, offering a faster, leaner alternative to Pandas UDFs for performant Python UDF execution in PySpark. By operating directly on Arrow data and eliminating the Pandas/Arrow conversion overhead, Arrow UDFs deliver ~10% faster execution, ~40% less memory usage, and better support for complex datatypes -- all with a familiar, intuitive decorator syntax.

Ready to explore more? Try out Native Arrow UDFs today on Databricks as part of Databricks Runtime 18.0. To get started, simply replace your existing Pandas UDFs with Arrow UDFs. In most cases, it only takes a few lines of change to unlock immediate performance gains. See the Arrow UDF documentation and Arrow UDTF documentation for the full API reference and additional examples.

Get the latest posts in your inbox

Subscribe to our blog and get the latest posts delivered to your inbox.

View all blogs