Pyspark：Concat 函数将列生成到新的数据框中答案

【问题标题】：Pyspark: Concat function generated columns into new dataframePyspark：Concat 函数将列生成到新的数据框中
【发布时间】：2018-11-10 18:42:48
【问题描述】：

我有一个带有 n 列的 pyspark 数据帧 (df)，我想生成另一个 n 列的 df，其中每列记录相应原始 df 列中连续行的百分比差异。并且新df中的列标题应该是==旧数据帧中的相应列标题+“_diff”。使用以下代码，我可以为原始 df 中的每一列生成新的百分比变化列，但无法将它们粘贴到具有合适列标题的新 df 中：

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
import pyspark.sql.functions as func

spark = (SparkSession
            .builder
            .appName('pct_change')
            .enableHiveSupport()
            .getOrCreate())

df = spark.createDataFrame([(1, 10, 11, 12), (2, 20, 22, 24), (3, 30, 33, 36)], 
                       ["index", "col1", "col2", "col3"])
w = Window.orderBy("index")

for i in range(1, len(df.columns)):
    col_pctChange = func.log(df[df.columns[i]]) - func.log(func.lag(df[df.columns[i]]).over(w))

谢谢

【问题讨论】：

标签： python dataframe pyspark

【解决方案1】：

在这种情况下，您可以在对 select 的调用中执行列表解析。

为了使代码更紧凑，我们可以先在一个列表中获取我们想要区分的列：

diff_columns = [c for c in df.columns if c != 'index']

接下来选择索引并遍历diff_columns 以计算新列。使用.alias() 重命名结果列：

df_diff = df.select(
    'index',
    *[(func.log(func.col(c)) - func.log(func.lag(func.col(c)).over(w))).alias(c + "_diff")
      for c in diff_columns]
)
df_diff.show()
#+-----+------------------+-------------------+-------------------+
#|index|         col1_diff|          col2_diff|          col3_diff|
#+-----+------------------+-------------------+-------------------+
#|    1|              null|               null|               null|
#|    2| 0.693147180559945| 0.6931471805599454| 0.6931471805599454|
#|    3|0.4054651081081646|0.40546510810816416|0.40546510810816416|
#+-----+------------------+-------------------+-------------------+

【讨论】：