使用 Spark 获取 reformat/shift 时间序列数据的有效方法

Question

我想使用 spark 构建一些时间序列模型。第一步是将序列数据重新格式化为训练样本。思路是：

原始时序数据（每个t*是一个数字）

t1  t2  t3  t4  t5  t6  t7  t8  t9  t10

期望的输出

t1  t2  t3  t4  t5  t6
t2  t3  t4  t5  t6  t7
t3  t4  t5  t6  t7  t8
..................

如何在 spark 中编写函数来执行此操作。函数签名应该像

重新格式化（数组[整数]，n：整数）

return 类型是 Dataframe 或 Vector

==========我在Spark 1.6.1上试过的代码=========

val arraydata=Array[Double](1,2,3,4,5,6,7,8,9,10)
val slideddata = arraydata.sliding(4).toSeq
val rows = arraydata.sliding(4).map{x=>Row(x:_*)}
sc.parallelize(arraydata.sliding(4).toSeq).toDF("Values")

最后一行无法通过错误：

Error:(52, 48) value toDF is not a member of org.apache.spark.rdd.RDD[Array[Double]]
    sc.parallelize(arraydata.sliding(4).toSeq).toDF("Values")

Answer 1

我无法弄清楚 n 的意义，因为它可以用作 window 大小以及它必须移动的值。

因此有两种口味：

如果 n 是 window 尺寸：

def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
sc.parallelize(arrayOfInteger.sliding(shiftValue).toSeq).toDF("values")
}

On REPL:

scala> def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
     | sc.parallelize(arrayOfInteger.sliding(shiftValue).toSeq).toDF("values")
     | }
reformat: (arrayOfInteger: Array[Int], shiftValue: Int)org.apache.spark.sql.DataFrame

scala> val arrayofInteger=(1 to 10).toArray
arrayofInteger: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> reformat(arrayofInteger,3).show
+----------+
|    values|
+----------+
| [1, 2, 3]|
| [2, 3, 4]|
| [3, 4, 5]|
| [4, 5, 6]|
| [5, 6, 7]|
| [6, 7, 8]|
| [7, 8, 9]|
|[8, 9, 10]|
+----------+

如果n是要移动的值：

def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
val slidingValue=arrayOfInteger.size-shiftValue
sc.parallelize(arrayOfInteger.sliding(slidingValue).toSeq).toDF("values")
}

On REPL:

scala> def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
     | val slidingValue=arrayOfInteger.size-shiftValue
     | sc.parallelize(arrayOfInteger.sliding(slidingValue).toSeq).toDF("values")
     | }
reformat: (arrayOfInteger: Array[Int], shiftValue: Int)org.apache.spark.sql.DataFrame

scala> val arrayofInteger=(1 to 10).toArray
arrayofInteger: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> reformat(arrayofInteger,3).show(false)
+----------------------+
|values                |
+----------------------+
|[1, 2, 3, 4, 5, 6, 7] |
|[2, 3, 4, 5, 6, 7, 8] |
|[3, 4, 5, 6, 7, 8, 9] |
|[4, 5, 6, 7, 8, 9, 10]|
+----------------------+

使用 Spark 获取 reformat/shift 时间序列数据的有效方法

efficient way to reformat/shift time series data using Spark

scala

time-series

apache-spark

spark-dataframe