SAS Proc Transpose到Pyspark

Question

我正在尝试将SAS Proc Transpose语句转换为Databricks中的Pyspark。 以以下数据为示例：

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}]

我希望结果看起来像这样，

我尝试使用pandas pivot_table（）函数，但我遇到了数据大小的某些性能问题：

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

是否有一种翻译方法Proc Transpose SAS逻辑到Pyspark而不是使用Pandas？

我正在尝试这样的事情，但是

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

如果您能帮助我，我会很感激！太感谢了。

Answer 1

我不知道您是如何从数据创建DF的，但这是我所做的：

import pyspark.pandas as ps

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

然后您的pandas代码工作了。

要使用Pyspark方法，这是我所做的：

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