SAS Proc Transpose to Pyspark

I am trying to convert a SAS proc transpose statement to pyspark in databricks. With the following data as a sample:

data = [{"duns":1234, "finc stress":100,"ver":6.0},{"duns":1234, "finc stress":125,"ver":7.0},{"duns":1234, "finc stress":135,"ver":7.1},{"duns":12345, "finc stress":125,"ver":7.6}]

I would expect the result to look like this

I tried using the pandas pivot_table() function with the following code however I ran into some performance issues with the size of the data:

tst = (df.pivot_table(index=['duns'], columns=['ver'], values='finc stress')
              .add_prefix('ver')
              .reset_index())

Is there a way to translate the PROC Transpose SAS logic to Pyspark instead of using pandas?

I am trying something like this but am getting an error

tst= sparkdf.groupBy('duns').pivot('ver').agg('finc_stress').withColumn('ver')

AssertionError: all exprs should be Column
---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
<command-2507760044487307> in <module>
      4 df = pd.DataFrame(data) # pandas
      5 
----> 6 tst= sparkdf.groupBy('duns').pivot('ver').agg('finc_stress').withColumn('ver')
      7 
      8 

/databricks/spark/python/pyspark/sql/group.py in agg(self, *exprs)
    115         else:
    116             # Columns
--> 117             assert all(isinstance(c, Column) for c in exprs), "all exprs should be Column"
    118             jdf = self._jgd.agg(exprs[0]._jc,
    119                                 _to_seq(self.sql_ctx._sc, [c._jc for c in exprs[1:]]))

AssertionError: all exprs should be Column

If you could help me out I would so appreciate it! Thank you so much.

Answer

I don’t know how you create df from data but here is what I did:

import pyspark.pandas as ps

df = ps.DataFrame(data)
df['ver'] = df['ver'].astype('str')

Then your pandas code worked.

To use PySpark method, here is what I did:

sparkdf.groupBy('duns').pivot('ver').agg(F.first('finc stress'))