Spark使用上一行的值向数据框添加新列

2 回答

• 29

  您可以使用 lag 窗口功能，如下所示

  from pyspark.sql.functions import lag, col
  from pyspark.sql.window import Window
  
  df = sc.parallelize([(4, 9.0), (3, 7.0), (2, 3.0), (1, 5.0)]).toDF(["id", "num"])
  w = Window().partitionBy().orderBy(col("id"))
  df.select("*", lag("num").over(w).alias("new_col")).na.drop().show()
  
  ## +---+---+-------+
  ## | id|num|new_col|
  ## +---+---+-------|
  ## |  2|3.0|    5.0|
  ## |  3|7.0|    3.0|
  ## |  4|9.0|    7.0|
  ## +---+---+-------+
  

  但是有一些重要的问题：

  • 如果您需要全局操作（未被其他一些列/列分区），则效率极低 .

  • 您需要一种自然的方式来订购数据 .

  虽然第二个问题几乎从来都不是问题，但第一个问题可能是一个交易破坏者 . 如果是这种情况，您只需将 DataFrame 转换为RDD并手动计算 lag . 参见例如：

  • How to transform data with sliding window over time series data in Pyspark

  • Apache Spark Moving Average（用Scala编写，但可以针对PySpark进行调整 . 请务必先阅读注释） .

  其他有用的链接：

  • https://github.com/UrbanInstitute/pyspark-tutorials/blob/master/05_moving-average-imputation.ipynb

  • Spark Window Functions - rangeBetween dates

  回复于 2024-04-30T22:07:50+08:00
• 0

  val df = sc.parallelize(Seq((4, 9.0), (3, 7.0), (2, 3.0), (1, 5.0))).toDF("id", "num")
  df.show
  +---+---+
  | id|num|
  +---+---+
  |  4|9.0|
  |  3|7.0|
  |  2|3.0|
  |  1|5.0|
  +---+---+
  df.withColumn("new_column", lag("num", 1, 0).over(w)).show
  +---+---+----------+
  | id|num|new_column|
  +---+---+----------+
  |  1|5.0|       0.0|
  |  2|3.0|       5.0|
  |  3|7.0|       3.0|
  |  4|9.0|       7.0|
  +---+---+----------+
  

  回复于 2024-04-30T22:07:50+08:00