spark sql窗口函数滞后答案

【问题标题】：spark sql window function lagspark sql窗口函数滞后
【发布时间】：2017-04-30 16:46:34
【问题描述】：

我正在查看 Scala 中 Spark DataFrame 的 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`

现在我想添加一个名称为(new_col) 的新列，其中一行向下滑动，如上所示。

我尝试了以下选项来使用窗口功能。

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

你有什么建议吗？

【问题讨论】：

@Ramesh 直到 Spark 2.0，用户必须使用 HiveContext 而不是 SQLContext 来应用窗口函数。通过传递SparkContext 的实例，HiveContext 的创建方式与SQLContext 相同。如果我没记错的话，您还需要包含 org.apache.spark:spark-hive_2.10 以及适合您的 Spark 发行版的版本。

标签： scala apache-spark window-functions

【解决方案1】：

你做得对，你错过的只是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 上运行

【讨论】：

我没有 1.5.2 设置，在我的机器中加载 1.5.2 (spark-hive jar) 时遇到问题。
我现在可以创建 hive 上下文了。但我仍然得到同样的错误。
我认为由于数据框是使用 sqlcontext 创建的，所以我仍然无法使用窗口函数。

【解决方案2】：

您可以导入以下两个包，这将解决延迟依赖的问题。

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

【讨论】：