["python","apache-spark","dataframe","pyspark","apache-spark-sql"]
Pyspark: Split multiple array columns into rows

I have a dataframe which has one row, and several columns. Some of the columns are single values, and others are lists. All list columns are the same length. I want to split each list column into a separate row, while keeping any non-list column as is. Sample DF: from pyspark import Row from pyspark.sql import SQLContext from pyspark.sql.functions import explode sqlc = SQLContext(sc) df = sqlc.createDataFrame([Row(a=1, b=[1,2,3],c=[7,8,9], d='foo')]) # +---+---------+---------+---+ # | a| b| c| d| # +---+---------+---------+---+ # | 1|[1, 2, 3]|[7, 8, 9]|foo| # +---+---------+---------+---+ What I want: +---+---+----+------+ | a| b| c | d | +---+---+----+------+ | 1| 1| 7 | foo | | 1| 2| 8 | foo | | 1| 3| 9 | foo | +---+---+----+------+ If I only had one list column, this would be easy by just doing an explode: df_exploded = df.withColumn('b', explode('b')) # >>> df_exploded.show() # +---+---+---------+---+ # | a| b| c| d| # +---+---+---------+---+ # | 1| 1|[7, 8, 9]|foo| # | 1| 2|[7, 8, 9]|foo| # | 1| 3|[7, 8, 9]|foo| # +---+---+---------+---+ However, if I try to also explode the c column, I end up with a dataframe with a length the square of what I want: df_exploded_again = df_exploded.withColumn('c', explode('c')) # >>> df_exploded_again.show() # +---+---+---+---+ # | a| b| c| d| # +---+---+---+---+ # | 1| 1| 7|foo| # | 1| 1| 8|foo| # | 1| 1| 9|foo| # | 1| 2| 7|foo| # | 1| 2| 8|foo| # | 1| 2| 9|foo| # | 1| 3| 7|foo| # | 1| 3| 8|foo| # | 1| 3| 9|foo| # +---+---+---+---+ What I want is - for each column, take the nth element of the array in that column and add that to a new row. I've tried mapping an explode accross all columns in the dataframe, but that doesn't seem to work either: df_split = df.rdd.map(lambda col: df.withColumn(col, explode(col))).toDF()
Spark >= 2.4 You can replace zip_ udf with arrays_zip function from pyspark.sql.functions import arrays_zip, col, explode (df .withColumn("tmp", arrays_zip("b", "c")) .withColumn("tmp", explode("tmp")) .select("a", col("tmp.b"), col("tmp.c"), "d")) Spark < 2.4 With DataFrames and UDF: from pyspark.sql.types import ArrayType, StructType, StructField, IntegerType from pyspark.sql.functions import col, udf, explode zip_ = udf( lambda x, y: list(zip(x, y)), ArrayType(StructType([ # Adjust types to reflect data types StructField("first", IntegerType()), StructField("second", IntegerType()) ])) ) (df .withColumn("tmp", zip_("b", "c")) # UDF output cannot be directly passed to explode .withColumn("tmp", explode("tmp")) .select("a", col("tmp.first").alias("b"), col("tmp.second").alias("c"), "d")) With RDDs: (df .rdd .flatMap(lambda row: [(row.a, b, c, row.d) for b, c in zip(row.b, row.c)]) .toDF(["a", "b", "c", "d"])) Both solutions are inefficient due to Python communication overhead. If data size is fixed you can do something like this: from functools import reduce from pyspark.sql import DataFrame # Length of array n = 3 # For legacy Python you'll need a separate function # in place of method accessor reduce( DataFrame.unionAll, (df.select("a", col("b").getItem(i), col("c").getItem(i), "d") for i in range(n)) ).toDF("a", "b", "c", "d") or even: from pyspark.sql.functions import array, struct # SQL level zip of arrays of known size # followed by explode tmp = explode(array(*[ struct(col("b").getItem(i).alias("b"), col("c").getItem(i).alias("c")) for i in range(n) ])) (df .withColumn("tmp", tmp) .select("a", col("tmp").getItem("b"), col("tmp").getItem("c"), "d")) This should be significantly faster compared to UDF or RDD. Generalized to support an arbitrary number of columns: # This uses keyword only arguments # If you use legacy Python you'll have to change signature # Body of the function can stay the same def zip_and_explode(*colnames, n): return explode(array(*[ struct(*[col(c).getItem(i).alias(c) for c in colnames]) for i in range(n) ])) df.withColumn("tmp", zip_and_explode("b", "c", n=3))