Convert columns to rows in Spark SQL

Question

I have some data like this:

ID	Value1	Value2	Value40
101	3	520	2001
102	29	530	2020

I want to take this data and convert in to a KV style pair instead

ID	ValueVv	ValueDesc
101	3	Value1
101	520	Value2
101	2001	Value40

I think it's a pivot, but I can't think of what this needs to look like in code.

I am trying to solve in PySQL but also in a Python DataFrame as I am using Spark.

I could easily, just union each column into an output using SQL, but I was hoping there is a more efficient way?

I've looked at melt as an option and stack. But I'm unsure how to do this effectively.

ZygD · Accepted Answer · 2022-05-03T08:11:01.657

It's the opposite of pivot - it's called unpivot.
In Spark, unpivoting is implemented using stack function.

Using PySpark, this is what you could do if you didn't have many columns:

from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame(
    [(101, 3, 520, 2001),
     (102, 29, 530, 2020)],
    ['ID', 'Value1', 'Value2', 'Value40'])

df = df.select(
    "ID",
    F.expr("stack(3, Value1, 'Value1', Value2, 'Value2', Value40, 'Value40') as (ValueVv, ValueDesc)")
)

From your example I see that you may have lots of columns. In this case you may use something like this:

cols_to_unpivot = [f"`{c}`, \'{c}\'" for c in df.columns if c != 'ID']
stack_string = ", ".join(cols_to_unpivot)
df = df.select(
    "ID",
    F.expr(f"stack({len(cols_to_unpivot)}, {stack_string}) as (ValueVv, ValueDesc)")
)

For the example data both versions return

+---+-------+---------+
| ID|ValueVv|ValueDesc|
+---+-------+---------+
|101|      3|   Value1|
|101|    520|   Value2|
|101|   2001|  Value40|
|102|     29|   Value1|
|102|    530|   Value2|
|102|   2020|  Value40|
+---+-------+---------+

This worked !! , I couldn't get the variable number of columns to work, as in my actual data i seem to have some datatype conversion errors. Im guessing a flow on issue which i will need to debug.. — Adam, Dec 23 '21 at 04:03
which i've just figured out is because the first column infers what the subsequent columns will be stacked into... in my case you can't stuff a string into an integer field.. — Adam, Dec 23 '21 at 05:03

score 0 · Answer 2 · answered Nov 29 '21 at 03:54

0

You can use flatmap as follows:

val schema = df.schema
val df2 = df.flatMap(row => {
    val id = row.getString(0)
    (1 until row.size).map(i => {
        (id, schema(i).name, row.getString(i))
    })
}).toDF("ID", "ValueVv", "ValueDesc")

df2.show()
+---+-------+---------+
| ID|ValueVv|ValueDesc|
+---+-------+---------+
|101| Value1|        3|
|101| Value2|      520|
|101|Value40|     2001|
|102| Value1|       29|
|102| Value2|      530|
|102|Value40|     2020|
+---+-------+---------+

or stack function from this link.

answered Nov 29 '21 at 03:54

Lamanus

12,898
4
21
47

Thank-you I will attempt to use this in my code.. Ill have another look at stack as well – Adam Nov 30 '21 at 06:26
This didn't work for me directly, instead i've done the following as an answer below – Adam Dec 14 '21 at 03:02

score 0 · Answer 3 · edited Dec 14 '21 at 05:45

I've done this so far to pivot, but wanting to make it happen not using pandas.. but using spark dataframe only.

import pandas as pd
    
def main():
    
    data={'AnID':[2001,2002,2003,2004],
          'Name':['adam','jane','Sarah','Ryan'], 
          'Age':[23,22,21,24], 
          'Age1':[24,52,51,264], 
          'Age2':[263,262,261,264]}

    df=pd.DataFrame(data)

   #Iterate the DataFrame so that we can pivot the "columns" into Rows
    schema = df.columns  #gives me the names of the columns
       
    df.index[0]
    
    #loop Through the id to pivot on (assume it's the first one in the df)
    
    j = 0
    df2=pd.DataFrame()
    while j < schema.size:
        curvalid = schema[j]
        idname = schema[j] #get each element of the array
        vval = df[idname].values  #Grab all the values for the given "column"
 
    #then get the data for that array element and populate a new object
    
        

        i = 0
        while i < vval.size:
            df3=pd.DataFrame({'DemoDesc' : curvalid, 'DemoID' : vval[i]}, index=[i])
            df2 = df2.append(df3,ignore_index=True)
            i = i + 1;
        j = j + 1;
    print(df2) #print the dataframe
    return;
    
main()

this is python code, my formatting-fu is terrible in SE :) – Adam Dec 14 '21 at 04:06 — Adam, Dec 14 '21 at 04:06

Convert columns to rows in Spark SQL

3 Answers3