How to return a “Tuple type” in a UDF in PySpark?

by

There is no such thing as a TupleType in Spark. Product types are represented as structs with fields of specific type. For example if you want to return an array of pairs (integer, string) you can use schema like this:

from pyspark.sql.types import *

schema = ArrayType(StructType([
    StructField("char", StringType(), False),
    StructField("count", IntegerType(), False)
]))

Example usage:

from pyspark.sql.functions import udf
from collections import Counter

char_count_udf = udf(
    lambda s: Counter(s).most_common(),
    schema
)

df = sc.parallelize([(1, "foo"), (2, "bar")]).toDF(["id", "value"])

df.select("*", char_count_udf(df["value"])).show(2, False)

## +---+-----+-------------------------+
## |id |value|PythonUDF#<lambda>(value)|
## +---+-----+-------------------------+
## |1  |foo  |[[o,2], [f,1]]           |
## |2  |bar  |[[r,1], [a,1], [b,1]]    |
## +---+-----+-------------------------+

More Related Contents:

  1. How to add a constant column in a Spark DataFrame?
  2. Spark Dataframe distinguish columns with duplicated name
  3. Pyspark: Split multiple array columns into rows
  4. How do I add a new column to a Spark DataFrame (using PySpark)?
  5. How to change a dataframe column from String type to Double type in PySpark?
  6. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  7. Retrieve top n in each group of a DataFrame in pyspark
  8. Updating a dataframe column in spark
  9. Dividing complex rows of dataframe to simple rows in Pyspark
  10. Filter Pyspark dataframe column with None value
  11. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  12. How to explode multiple columns of a dataframe in pyspark
  13. Spark add new column to dataframe with value from previous row
  14. PySpark: multiple conditions in when clause
  15. Pyspark: Replacing value in a column by searching a dictionary
  16. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  17. Create Spark DataFrame. Can not infer schema for type
  18. Pivot String column on Pyspark Dataframe
  19. Filtering DataFrame using the length of a column
  20. Multiple condition filter on dataframe
  21. How to change dataframe column names in pyspark?
  22. How to export a table dataframe in PySpark to csv?
  23. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  24. Why is Apache-Spark – Python so slow locally as compared to pandas?
  25. PySpark create new column with mapping from a dict
  26. Using UDF ignores condition in when
  27. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  28. Concatenate two PySpark dataframes
  29. How to drop all columns with null values in a PySpark DataFrame?
  30. Pyspark changing type of column from date to string

Leave a Comment