代码之家 › 专栏 › 技术社区 › jingx

如何计算`值[i]-值[i-1]`火花法?[副本]

apache-spark-sql apache-spark

jingx · 技术社区 · 8 年前

window 火花滑动功能 DataFrame 在斯卡拉。

数据帧 带列 Col1 Col2 Col3 , date volume 和 new_col .

Col1    Col2    Col3    date     volume new_col
                        201601  100.5   
                        201602  120.6   100.5
                        201603  450.2   120.6
                        201604  200.7   450.2
                        201605  121.4   200.7`

新颜色

val windSldBrdrxNrx_df = df.withColumn("Prev_brand_rx", lag("Prev_brand_rx",1))

你有什么建议吗?

0 回复 | 直到 6 年前

mrsrinivas 8 年前

你做得对你错过的就是 over(window expression) 在 lag

val df = sc.parallelize(Seq((201601, 100.5),
  (201602, 120.6),
  (201603, 450.2),
  (201604, 200.7),
  (201605, 121.4))).toDF("date", "volume")

val w = org.apache.spark.sql.expressions.Window.orderBy("date")  

import org.apache.spark.sql.functions.lag

val leadDf = df.withColumn("new_col", lag("volume", 1, 0).over(w))

leadDf.show()

+------+------+-------+
|  date|volume|new_col|
+------+------+-------+
|201601| 100.5|    0.0|
|201602| 120.6|  100.5|
|201603| 450.2|  120.6|
|201604| 200.7|  450.2|
|201605| 121.4|  200.7|
+------+------+-------+

此代码在Spark shell 2.0.2上运行

Sampat Kumar 9 年前

您可以导入以下两个包,这将解决滞后依赖性的问题。

import org.apache.spark.sql.functions.{lead, lag}
import org.apache.spark.sql.expressions.Window

推荐文章

maximodesousadias · 如何根据条件删除日期后的记录

2 年前

Joe Bloggr · 如何将Dataframe类型的函数参数传递给SparkSQL查询

2 年前

Shankar Panda · 如何从org.apache.spark.sql获取密钥。在scala中键入列并将其放入列表变量中?

2 年前

Aaron Brazier · 连接2个pyspark数据帧并继续运行窗口sum和max

2 年前

user23358051 · 火花顺序优化规则

2 年前

Tristpost · 如何从我自己的Java Stream将CSV数据最好地加载到Apache Spark数据帧中?

2 年前

David Cunningham · Pyspark结构化流媒体-来自以前记录的数据

2 年前

Surender Raja · 对case语句的pyspark查询引发错误

2 年前

ConfusedDeveloper · 取消查看SPARK SQL中的列组

3 年前

nfsp412 · 当我使用spark-sql时,将出现此错误

3 年前