How to zip two array columns in Spark SQL

by

A Spark SQL equivalent of Python’s would be pyspark.sql.functions.arrays_zip:

pyspark.sql.functions.arrays_zip(*cols)

Collection function: Returns a merged array of structs in which the N-th struct contains all N-th values of input arrays.

So if you already have two arrays:

from pyspark.sql.functions import split

df = (spark
    .createDataFrame([('abc, def, ghi', '1.0, 2.0, 3.0')])
    .toDF("column_1", "column_2")
    .withColumn("column_1", split("column_1", "\s*,\s*"))
    .withColumn("column_2", split("column_2", "\s*,\s*")))

You can just apply it on the result

from pyspark.sql.functions import arrays_zip

df_zipped = df.withColumn(
  "zipped", arrays_zip("column_1", "column_2")
)

df_zipped.select("zipped").show(truncate=False)

+------------------------------------+
|zipped                              |
+------------------------------------+
|[[abc, 1.0], [def, 2.0], [ghi, 3.0]]|
+------------------------------------+

Now to combine the results you can transform (How to use transform higher-order function?, TypeError: Column is not iterable – How to iterate over ArrayType()?):

df_zipped_concat = df_zipped.withColumn(
    "zipped_concat",
     expr("transform(zipped, x -> concat_ws('_', x.column_1, x.column_2))")
) 

df_zipped_concat.select("zipped_concat").show(truncate=False)

+---------------------------+
|zipped_concat              |
+---------------------------+
|[abc_1.0, def_2.0, ghi_3.0]|
+---------------------------+

Note:

Higher order functions transform and arrays_zip has been introduced in Apache Spark 2.4.

More Related Contents:

  1. Why is Apache-Spark – Python so slow locally as compared to pandas?
  2. Convert pyspark string to date format
  3. How to split Vector into columns – using PySpark
  4. How to use JDBC source to write and read data in (Py)Spark?
  5. Applying UDFs on GroupedData in PySpark (with functioning python example)
  6. Pyspark: Split multiple array columns into rows
  7. How to change a dataframe column from String type to Double type in PySpark?
  8. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  9. Retrieve top n in each group of a DataFrame in pyspark
  10. Pyspark: explode json in column to multiple columns
  11. Spark DataFrame: Computing row-wise mean (or any aggregate operation)
  12. How to use a Scala class inside Pyspark
  13. Filter Pyspark dataframe column with None value
  14. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  15. Cast column containing multiple string date formats to DateTime in Spark
  16. How to explode multiple columns of a dataframe in pyspark
  17. PySpark: multiple conditions in when clause
  18. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  19. Load CSV file with PySpark
  20. Reshaping/Pivoting data in Spark RDD and/or Spark DataFrames
  21. Cannot find col function in pyspark
  22. What is the best way to remove accents with Apache Spark dataframes in PySpark?
  23. How to pass a constant value to Python UDF?
  24. How do I convert an array (i.e. list) column to Vector
  25. Add column sum as new column in PySpark dataframe
  26. When to cache a DataFrame?
  27. How to add suffix and prefix to all columns in python/pyspark dataframe
  28. How to drop all columns with null values in a PySpark DataFrame?
  29. Pyspark changing type of column from date to string
  30. PySpark: compute row maximum of the subset of columns and add to an exisiting dataframe

Leave a Comment